为什么 Spark/Scala 编译器无法在 RDD[Map[Int, Int]] 上找到 toDF？答案

【问题标题】：Why does Spark/Scala compiler fail to find toDF on RDD[Map[Int, Int]]?为什么 Spark/Scala 编译器无法在 RDD[Map[Int, Int]] 上找到 toDF？
【发布时间】：2015-12-07 23:36:52
【问题描述】：

为什么下面会报错？

scala> import sqlContext.implicits._
import sqlContext.implicits._

scala> val rdd = sc.parallelize(1 to 10).map(x => (Map(x  -> 0), 0))
rdd: org.apache.spark.rdd.RDD[(scala.collection.immutable.Map[Int,Int], Int)] = MapPartitionsRDD[20] at map at <console>:27

scala> rdd.toDF
res8: org.apache.spark.sql.DataFrame = [_1: map<int,int>, _2: int]

scala> val rdd = sc.parallelize(1 to 10).map(x => Map(x  -> 0))
rdd: org.apache.spark.rdd.RDD[scala.collection.immutable.Map[Int,Int]] = MapPartitionsRDD[23] at map at <console>:27

scala> rdd.toDF
<console>:30: error: value toDF is not a member of org.apache.spark.rdd.RDD[scala.collection.immutable.Map[Int,Int]]
              rdd.toDF

那么这里到底发生了什么，toDF 可以将(scala.collection.immutable.Map[Int,Int], Int) 类型的RDD 转换为DataFrame，但不能将scala.collection.immutable.Map[Int,Int] 类型。这是为什么？

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

出于同样的原因，你不能使用

sqlContext.createDataFrame(1 to 10).map(x => Map(x  -> 0))

如果您查看org.apache.spark.sql.SQLContext 源代码，您会发现createDataFrame 方法的两种不同实现：

def createDataFrame[A <: Product : TypeTag](rdd: RDD[A]): DataFrame

和

def createDataFrame[A <: Product : TypeTag](data: Seq[A]): DataFrame

如您所见，两者都要求 A 是 Product 的子类。当您在RDD[(Map[Int,Int], Int)] 上调用toDF 时，它会起作用，因为Tuple2 确实是Product。 Map[Int,Int] 本身并不是错误。

您可以通过将Map 与Tuple1 包装起来使其工作：

sc.parallelize(1 to 10).map(x => Tuple1(Map(x  -> 0))).toDF

【讨论】：

【解决方案2】：

基本上是因为没有隐式为 RDD 内的 Map 创建 DataFrame。

在您的第一个示例中，您将返回一个元组，它是一个具有隐式转换的产品。

rddToDataFrameHolder[A <: Product : TypeTag](rdd: RDD[A])

在第二个示例中，您使用的 RDD 中有一个 Map，它没有隐式转换。

【讨论】：