createDataFrame() выдает исключение при передаче javaRDD, содержащего столбец ArrayType в SPARK 2.1.

Я хочу создать Dataframe (он же Dataset‹Row› в Spark 2.1) с помощью createDataframe(), все работает хорошо, когда я передаю ему параметр List‹Row›, но он выдает исключение, когда я передаю JavaRDD‹Row›.

[Код]

SparkSession ss = SparkSession.builder().appName("Spark Test").master("local[4]").getOrCreate();

List<Row> data = Arrays.asList(
        RowFactory.create(Arrays.asList("a", "b", "c")),
        RowFactory.create(Arrays.asList("A", "B", "C"))
);

StructType schema = new StructType(new StructField[]{
        DataTypes.createStructField("col_1", DataTypes.createArrayType(DataTypes.StringType), false)
});

Когда я пробую этот код, все работает хорошо

ss.createDataFrame(data, schema).show();

+---------+
|    col_1|
+---------+
|[a, b, c]|
|[A, B, C]|
+---------+

Но когда я передаю JavaRDD в качестве первого параметра, возникает исключение

JavaRDD<Row> rdd = JavaSparkContext.fromSparkContext(ss.sparkContext()).parallelize(data);

ss.createDataFrame(rdd, schema).show(); // throws exception

[Исключение]

Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 2 in stage 1.0 failed 1 times, most recent failure: Lost task 2.0 in stage 1.0 (TID 3, localhost, executor driver): java.lang.RuntimeException: Error while encoding: java.lang.RuntimeException: java.util.Arrays$ArrayList is not a valid external type for schema of array<string>
mapobjects(MapObjects_loopValue0, MapObjects_loopIsNull1, ObjectType(class java.lang.Object), staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(lambdavariable(MapObjects_loopValue0, MapObjects_loopIsNull1, ObjectType(class java.lang.Object)), StringType), true), validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object), 0, col_1), ArrayType(StringType,true))) AS col_1#0
+- mapobjects(MapObjects_loopValue0, MapObjects_loopIsNull1, ObjectType(class java.lang.Object), staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(lambdavariable(MapObjects_loopValue0, MapObjects_loopIsNull1, ObjectType(class java.lang.Object)), StringType), true), validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object), 0, col_1), ArrayType(StringType,true)))
   :- staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(lambdavariable(MapObjects_loopValue0, MapObjects_loopIsNull1, ObjectType(class java.lang.Object)), StringType), true)
   :  +- validateexternaltype(lambdavariable(MapObjects_loopValue0, MapObjects_loopIsNull1, ObjectType(class java.lang.Object)), StringType)
   :     +- lambdavariable(MapObjects_loopValue0, MapObjects_loopIsNull1, ObjectType(class java.lang.Object))
   +- validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object), 0, col_1), ArrayType(StringType,true))
      +- getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object), 0, col_1)
         +- assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object)
            +- input[0, org.apache.spark.sql.Row, true]

    at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.toRow(ExpressionEncoder.scala:293)
    at org.apache.spark.sql.SparkSession$$anonfun$3.apply(SparkSession.scala:547)
    at org.apache.spark.sql.SparkSession$$anonfun$3.apply(SparkSession.scala:547)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:409)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:232)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$2.apply(SparkPlan.scala:225)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:826)
    at org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$25.apply(RDD.scala:826)
    at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38)
    at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:323)
    at org.apache.spark.rdd.RDD.iterator(RDD.scala:287)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:99)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:282)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.RuntimeException: java.util.Arrays$ArrayList is not a valid external type for schema of array<string>
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source)
    at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.toRow(ExpressionEncoder.scala:290)
    ... 17 more

Любая помощь будет оценена


java apache-spark apache-spark-dataset apache-spark-2.0 spark-dataframe
person FelixHo    schedule 14.04.2017    source источник
comment
если вы попробуете rdd.toDF, у вас будет такая же ошибка?   -  person Mehrez    schedule 14.04.2017

Ответы (1)


arrow_upward
0
arrow_downward

Причина, по которой вы сталкиваетесь с этой проблемой, заключается в том, что Spark не может преобразовать ArrayList в тип String[]

Я бы изменил строки для создания типов String[]. Попробуйте этот код:

List<Row> data = Arrays.asList(
        Arrays.asList("a", "b", "c"),
        Arrays.asList("A", "B", "C")
).stream().map(r -> {
   String[] arr = r.toArray(new String[r.size()]);
   return RowFactory.create( new Object[]{arr});
}).collect(Collectors.toList());
person webber    schedule 16.01.2019