Spark RDD 到 Dataframe 并指定模式答案

【问题标题】：Spark RDD to Dataframe with schema specifyingSpark RDD 到 Dataframe 并指定模式
【发布时间】：2015-09-16 11:32:27
【问题描述】：

在通过 Row 对象从 RDD 转换 DataFrame 时，spark 似乎无法为 DataFrame 应用模式（不同于 String）。我在 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 版本中标记为已解决。

【问题讨论】：

标签： apache-spark dataframe apache-spark-sql

【解决方案1】：

您正在混合两种不同的东西 - 数据类型和 DataFrame 架构。当您像这样创建Row 时：

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

你得到Row(_: String, _: String)，但你的架构表明你有Row(_: Integer, _: String)。由于没有自动类型转换，因此您会收到错误。

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

【讨论】：

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