在Spark/Scala中将Array[Row]转换为DataFrame

Question

我想将Array[org.apache.spark.sql.Row]转换为DataFrame，请问有更好的方法吗？

我试过先将它转换为RDD，然后再转换成Dataframe，但当我对DataFrame进行任何操作时，会出现异常。

val arrayOfRows = myDataFrame.collect().map(t => myfun(t))
val distDataRDD = sc.parallelize(arrayOfRows)
val newDataframe = sqlContext.createDataFrame(distDataRDD,myschema)

这里的myfun()是一个函数，它返回Row(org.apache.spark.sql.Row)。

数组中的内容是正确的，我能够打印出来而没有任何问题。

但是当我尝试对RDD中的记录进行计数时，它给了我计数以及一个警告，提示其中一个阶段包含了一个非常大的任务。我想我可能在做什么错误的事情。请帮忙解决。

- rvp

Answer 1

第一行有一个错误。`collect` 返回一个数组，而 `map` 是在 DataFrame/RDD 上操作的方法。

尝试使用 `val arrayOfRows = myDataFrame.map(t => myfun(t)).collect()`。

- Hussein Hazimeh

---

当我改变顺序时，出现了以下错误：org.apache.spark.SparkException: Task not serializable。 - rvp

arrayofRows 实际上是 DataFrame 类型，因此无需第 2 和第 3 行（sc.parallelize 接受 RDD 而不是 DataFrame，这是新异常的原因）。 - Hussein Hazimeh

一进入第一行 val arrayOfRows = myDataFrame.collect().map(t => myfun(t))，我就收到了错误提示。 - rvp

Answer 2

case class PgRnk (userId : Long , pageRank: Double ) 
// create a case class 

sc.parallelize(pg10.map(r1 => PgRnk(r1.getLong(0), r1.getDouble(1)))).toDS() 
// convert into a dataset, sc.parallelize converts the array into a RDD, and then to DS 

- Ashish Markanday