Добавление нового столбца в DataFrame с литеральным значением типа set

Question

Map<File, Dataset<Row> allWords = ...
StructField[] structFields = new StructField[] {
        new StructField("word", DataTypes.StringType, false, Metadata.empty()),
        new StructField("count", DataTypes.IntegerType, false, Metadata.empty()),
        new StructField("files", ???, false, Metadata.empty())
};
StructType structType = new StructType(structFields);

Dataset<Row> allFilesWords = spark.createDataFrame(new ArrayList<>(), structType);

for (Map.Entry<File, Dataset<Row>> entry : allWords.entrySet()) {
    Integer fileIndex = files.indexOf(entry.getKey());
    allFilesWords.unionAll(
            allWords.get(entry.getKey()).withColumn("files", ???)
    );
}

В приведенном выше коде allWords представляет сопоставление файла с количеством слов (Row: (string, integer)). Теперь я хочу объединить результат для всех файлов в один DataFrame, сохранив исходный файл, в котором было упомянуто слово. Так как в конце каждое слово могло быть упомянуто в нескольких файлах, столбец files имеет тип набор целых чисел (при условии, что файлы отображаются в целые числа). Теперь я пытаюсь добавить новый столбец в allWords фреймы данных, а затем использовать unionAll, чтобы объединить их все вместе.

Но я не знаю, как определить и инициализировать новый столбец (здесь он называется files) с набором, содержащим только один элемент fileIndex.

Благодаря ссылке, приведенной в комментариях, я знаю, что должен использовать functions.typedLit, но эта функция запрашивает второй параметр, который я не знаю, что для него предоставить. Кроме того, я не знаю, как определить столбец. И последнее, предоставленная ссылка на Python, и я ищу Java API.

Answer 1

Я сам нашел решение (с некоторой помощью):

Map<File, Dataset<Row> allWords = ...
StructField[] structFields = new StructField[] {
        new StructField("word", DataTypes.StringType, false, Metadata.empty()),
        new StructField("count", DataTypes.IntegerType, false, Metadata.empty()),
        new StructField("files", DataTypes.createArrayType(DataTypes.IntegerType), true, Metadata.empty())
};
StructType structType = new StructType(structFields);

Dataset<Row> allFilesWords = spark.createDataFrame(new ArrayList<>(), structType);
for (Map.Entry<File, Dataset<Row>> entry : allWords.entrySet()) {
    Integer fileIndex = files.indexOf(entry.getKey());
    allFilesWords.unionAll(
            allWords.get(entry.getKey())
                    .withColumn("files", functions.typedLit(seq, MyTypeTags.SeqInteger()))
    );
}

Проблема заключалась в том, что TypeTag - это артефакт времени компиляции из Scala, основанный на том, что я получил в этом другой вопрос , он должен быть сгенерирован компилятором Scala, и вы не сможете сгенерировать его в Java.Итак, мне пришлось скомпоновать мою пользовательскую структуру данных TypeTag в файле Scala и добавить ее в мой проект Java Maven.Для этого я следовал этой статье .

А вот мой MyTypeTags.scala файл:

import scala.reflect.runtime.universe._

object MyTypeTags {
  val SeqInteger = typeTag[Seq[Integer]]
}