如何将基于case class的RDD转换为DataFrame？

Question

Spark文档演示了如何使用Scala case类推断模式从RDD创建DataFrame。我尝试使用sqlContext.createDataFrame(RDD, CaseClass)来复制这个概念，但我的DataFrame最终为空。下面是我的Scala代码：

// sc is the SparkContext, while sqlContext is the SQLContext.

// Define the case class and raw data
case class Dog(name: String)
val data = Array(
    Dog("Rex"),
    Dog("Fido")
)

// Create an RDD from the raw data
val dogRDD = sc.parallelize(data)

// Print the RDD for debugging (this works, shows 2 dogs)
dogRDD.collect().foreach(println)

// Create a DataFrame from the RDD
val dogDF = sqlContext.createDataFrame(dogRDD, classOf[Dog])

// Print the DataFrame for debugging (this fails, shows 0 dogs)
dogDF.show()

我看到的输出是：

Dog(Rex)
Dog(Fido)
++
||
++
||
||
++

我错过了什么？

谢谢！

- sparkour

Answer 1

你所需要的只是

val dogDF = sqlContext.createDataFrame(dogRDD)

第二个参数是 Java API 的一部分，它期望你的类遵循 Java Beans 约定（有 getters/setters 函数）。由于您的 case class 未遵循这个约定，因此没有检测到任何属性，导致生成的 DataFrame 没有列。

- Vitalii Kotliarenko

---

这个方法可行。我还不得不将 case class 的定义移到主函数之外，以避免出现“没有可用于 Dog 的 TypeTag”错误。谢谢！ - sparkour

我明白了，非常有趣。因此，只有在从Java API调用时才需要第二个参数，Scala会自动检测应转换为列的类型字段？ - qwwqwwq

只有在将 case class 移到 main 之外时才能正常工作。@Vitalii，@sparkour...是否有任何解释为什么需要将 case class 移到 main 之外。 - Praveen L

我的 case class 中有一个字段名为 abstract，因此我收到了“abstract 是保留关键字，不能用作字段名”的错误提示。有什么解决方法吗？ - Anish

Answer 2

您可以通过以下方式使用 toDF 直接从 Seq 的实例列表创建 DataFrame ：

val dogDf = Seq(Dog("Rex"), Dog("Fido")).toDF

- David Griffin

---

记得添加 import spark.implicits._ - Federico Rizzo

Answer 3

在集群模式下，Case Class 方法不起作用。它会给你定义的 Case Class 抛出 ClassNotFoundException 异常。

将其转换为 RDD[Row]，并使用 StructField 定义您的 RDD 的模式，然后像这样 createDataFrame

val rdd = data.map { attrs => Row(attrs(0),attrs(1)) }  

val rddStruct = new StructType(Array(StructField("id", StringType, nullable = true),StructField("pos", StringType, nullable = true)))

sqlContext.createDataFrame(rdd,rddStruct)

toDF() 也不起作用

- Kamaldeep Singh

---

为什么它在集群模式下无法工作？ - Remis Haroon - رامز