将文字值设置为set的新列添加到DataFrame中

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列设计为整数类型集（假设文件被映射为整数）。现在，我正在尝试向allWordsDataFrames添加一个新列，然后使用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文件中编写我的自定义数据结构，并将其添加到我的Maven Java项目中。为此，我关注了本文。

这是我的MyTypeTags.scala文件：

import scala.reflect.runtime.universe._

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