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

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: Error while encoding: java.util.Arrays$ArrayList is not a valid external type for schema of array

发生在 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>到数据框。我相信 Spark Java API 有一个重载方法 createDataFrame() ，它接受 JavaRDD<Row>并根据提供的模式将其转换为 DataFrame。我很困惑为什么它不起作用。有人可以帮忙吗？

Answer 1

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

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

java.util.Arrays$ArrayList is not a valid external type for schema of array

因此，直接传递 String[] 应该有效:

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

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

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

但是，从 Spark 文档中的 Java 行列表构造 DataFrame 仍然是个谜。这是因为 SparkSession.createDataFrame 函数版本以行的第一个参数 java.util.List 进行特殊的类型检查和转换，以便将所有 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);

希望这能解决您的问题。