将 JavaRDD 转换为 DataFrame 时出现 Spark 错误：java.util.Arrays$ArrayList 不是 array<string> 架构的有效外部类型

Question

我使用的是 Spark 2.1.0。对于以下代码，它读取文本文件并将内容转换为 DataFrame，然后输入 Word2Vector 模型：

SparkSession spark = SparkSession.builder().appName("word2vector").getOrCreate();
JavaRDD<String> lines = spark.sparkContext().textFile("input.txt", 10).toJavaRDD();
JavaRDD<List<String>> lists = lines.map(new Function<String, List<String>>(){
                public List<String> call(String line){
                    List<String> list = Arrays.asList(line.split(" "));
                    return list;
                }
            });

JavaRDD<Row> rows = lists.map(new Function<List<String>, Row>() {
                public Row call(List<String> list) {
                    return RowFactory.create(list);
                }
            });


StructType schema = new StructType(new StructField[] {
                        new StructField("text", new ArrayType(DataTypes.StringType, true), false, Metadata.empty()) 
                    });

Dataset<Row> input = spark.createDataFrame(rows, schema);
input.show(3);
Word2Vec word2Vec = new Word2Vec().setInputCol("text").setOutputCol("result").setVectorSize(100).setMinCount(0);
Word2VecModel model = word2Vec.fit(input);
Dataset<Row> result = model.transform(input);

抛出异常

java.lang.RuntimeException：编码时出错：java.util.Arrays$ArrayList 不是有效的外部类型 数组架构

这发生在 input.show(3) 行，因此 createDataFrame() 导致异常，因为 Arrays.asList() 返回一个 Arrays$ArrayList 此处不支持。但是 Spark 官方文档有以下代码：

List<Row> data = Arrays.asList(
      RowFactory.create(Arrays.asList("Hi I heard about Spark".split(" "))),
      RowFactory.create(Arrays.asList("I wish Java could use case classes".split(" "))),
      RowFactory.create(Arrays.asList("Logistic regression models are neat".split(" ")))
    );

StructType schema = new StructType(new StructField[]{
      new StructField("text", new ArrayType(DataTypes.StringType, true), false, Metadata.empty())
});
Dataset<Row> documentDF = spark.createDataFrame(data, schema);

效果很好。如果 Arrays$ArrayList 不受支持，那么这段代码是如何工作的？不同之处在于我将JavaRDD<Row> 转换为DataFrame，但官方文档将List<Row> 转换为DataFrame。我相信 Spark Java API 有一个重载方法 createDataFrame()，它接受 JavaRDD<Row> 并根据提供的模式将其转换为 DataFrame。我很困惑为什么它不起作用。任何人都可以帮忙吗？

Answer 1

几天前我遇到了同样的问题，解决这个问题的唯一方法是使用数组数组。为什么 ？回复如下：

ArrayType 是 Scala 数组的包装器，它与 Java 数组一一对应。默认情况下，Java ArrayList 未映射到 Scala Array，这就是您得到异常的原因：

java.util.Arrays$ArrayList 不是数组架构的有效外部类型

因此，直接传递一个 String[] 是可行的：

RowFactory.create(line.split(" "))

但由于 create 将对象列表作为输入，因为一行可能有一个列列表，因此 String[] 被解释为字符串列的列表。这就是为什么需要一个双字符串数组的原因：

RowFactory.create(new String[][] {line.split(" ")})

然而，从 spark 文档中的 Java 行列表构造 DataFrame 仍然是个谜。这是因为作为第一个参数 java.util.List 行的 SparkSession.createDataFrame 函数版本会进行特殊的类型检查和转换，以便将所有 Java Iterable（即 ArrayList）转换为 Scala 数组。 但是，采用 JavaRDD 的 SparkSession.createDataFrame 直接将行内容映射到 DataFrame。

总结一下，这是正确的版本：

    SparkSession spark = SparkSession.builder().master("local[*]").appName("Word2Vec").getOrCreate();
    SparkContext sc = spark.sparkContext();
    sc.setLogLevel("WARN");
    JavaRDD<String> lines = sc.textFile("input.txt", 10).toJavaRDD();
    JavaRDD<Row> rows = lines.map(new Function<String, Row>(){
        public Row call(String line){
            return RowFactory.create(new String[][] {line.split(" ")});
        }
    });

    StructType schema = new StructType(new StructField[] {
            new StructField("text", new ArrayType(DataTypes.StringType, true), false, Metadata.empty())
    });
    Dataset<Row> input = spark.createDataFrame(rows, schema);
    input.show(3);

希望这能解决您的问题。

Answer 2

正如错误所说。 ArrayList 不等同于 Scala 的数组。您应该改用普通数组（即String[]）。

Answer 3

对我来说，下面的工作正常

JavaRDD<Row> rowRdd = rdd.map(r -> RowFactory.create(r.split(",")));