使用指定架构将Spark RDD转换为Dataframe

Question

似乎在将RDD通过Row对象转换为DataFrame时，Spark无法应用模式（与字符串不同）。我已经尝试了Spark 1.4和1.5版本。

片段（Java API）:

JavaPairInputDStream<String, String> directKafkaStream = KafkaUtils.createDirectStream(jssc, String.class, String.class,
                StringDecoder.class, StringDecoder.class, kafkaParams, topicsSet);

directKafkaStream.foreachRDD(rdd -> {
    rdd.foreach(x -> System.out.println("x._1() = " + x._1()));
    rdd.foreach(x -> System.out.println("x._2() = " + x._2()));

    JavaRDD<Row> rowRdd = rdd.map(x -> RowFactory.create(x._2().split("\t")));

    rowRdd.foreach(x -> System.out.println("x = " + x));

    SQLContext sqlContext = SQLContext.getOrCreate(rdd.context());

    StructField id = DataTypes.createStructField("id", DataTypes.IntegerType, true);
    StructField name = DataTypes.createStructField("name", DataTypes.StringType, true);
    List<StructField> fields = Arrays.asList(id, name);
    StructType schema = DataTypes.createStructType(fields);

    DataFrame sampleDf = sqlContext.createDataFrame(rowRdd, schema);

    sampleDf.printSchema();
    sampleDf.show();

    return null;
});

jssc.start();
jssc.awaitTermination();

如果为"id"字段指定DataTypes.StringType，则会产生以下输出：

x._1() = null
x._2() = 1  item1
x = [1,item1]
root
 |-- id: string (nullable = true)
 |-- name: string (nullable = true)

+---+-----+
| id| name|
+---+-----+
|  1|item1|
+---+-----+

特定的代码会抛出错误：

x._1() = null
x._2() = 1  item1
x = [1,item1]
root
 |-- id: integer (nullable = true)
 |-- name: string (nullable = true)

15/09/16 04:13:33 ERROR JobScheduler: Error running job streaming job 1442402013000 ms.0
java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Integer
    at scala.runtime.BoxesRunTime.unboxToInt(BoxesRunTime.java:106)
    at org.apache.spark.sql.catalyst.expressions.BaseGenericInternalRow$class.getInt(rows.scala:40)
    at org.apache.spark.sql.catalyst.expressions.GenericInternalRow.getInt(rows.scala:220)
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$IntConverter$.toScalaImpl(CatalystTypeConverters.scala:358)

类似的问题在Spark Confluence上出现过，但在1.3版本中被标记为已解决。

- morsik

Answer 1

您正在混淆两个不同的概念 - 数据类型和DataFrame模式。当您创建Row时，就像这样：

RowFactory.create(x._2().split("\t"))

你得到了Row(_: String, _: String)，但是你的模式表明你有Row(_: Integer, _: String)。由于没有自动类型转换，所以你会得到错误。

要使它工作，你可以在创建行时进行值转换，或将id定义为StringType，并在创建DataFrame后使用Column.cast方法。

- zero323

---

是的，我认为 Spark 在创建 DataFrame 时将模式应用于行。谢谢。 - morsik