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

Question

为什么下面的结果会出错？

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 类型转换为 DataFrame。 collection.immutable.Map[Int,Int]。这是为什么？

Answer 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 确实是一个 产品。 Map[Int,Int] 本身并不是错误。

您可以通过用 Tuple1 包装 Map 来使其工作:

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