在 Spark-Scala 中将 List 或 RDD 的列表转换为 DataFrame答案

【问题标题】：Converting List of List or RDD to DataFrame in Spark-Scala在 Spark-Scala 中将 List 或 RDD 的列表转换为 DataFrame
【发布时间】：2019-08-12 06:06:47
【问题描述】：

所以基本上我想要实现的是 - 我有一个有 4 列的表（比如说），我将它暴露给一个 DataFrame - DF1。现在我想将 DF1 的每一行存储到另一个配置单元表（基本上是 DF2，其架构为 - Column1、Column2、Column3），而 column3 值将是 DataFrame DF1 的“-”分隔行。

val df = hiveContext.sql("from hive_table SELECT *")
val writeToHiveDf = df.filter(new Column("id").isNotNull)

var builder : List[(String, String, String)] = Nil
    var finalOne  =  new ListBuffer[List[(String, String, String)]]()
    writeToHiveDf.rdd.collect().foreach {
      row =>
        val item = row.mkString("-@")
        builder = List(List("dummy", "NEVER_NULL_CONSTRAINT", "some alpha")).map{case List(a,b,c) => (a,b,c)}
        finalOne += builder
    }

现在我将 finalOne 作为列表列表，我想直接或通过 RDD 将其转换为数据帧。

var listRDD = sc.parallelize(finalOne) //Converts to RDD - It works. 
val dataFrameForHive : DataFrame = listRDD.toDF("table_name", "constraint_applied", "data") //Doesn't work

错误：

java.lang.ClassCastException: org.apache.spark.sql.types.ArrayType cannot be cast to org.apache.spark.sql.types.StructType
    at org.apache.spark.sql.SQLContext.createDataFrame(SQLContext.scala:414)
    at org.apache.spark.sql.SQLImplicits.rddToDataFrameHolder(SQLImplicits.scala:94)

有人可以帮助我理解将其转换为 DataFrame 的正确方法吗？非常感谢您的支持。

【问题讨论】：

您希望数据框具有什么模式，3 列字符串类型或 1 列元素为结构（3 个字符串）的数组类型？

标签： scala exception casting apache-spark-sql rdd

【解决方案1】：

如果您希望数据框中包含 3 列字符串类型，则应将 List[List[(String,String,String)]] 展平为 List[(String,String,String)]：

var listRDD = sc.parallelize(finalOne.flatten) // makes List[(String,String,String)]
val dataFrameForHive : DataFrame = listRDD.toDF("table_name", "constraint_applied", "data")

【讨论】：

【解决方案2】：

我相信在将“finalOne”数据帧传递给 sc.parallelize() 函数之前将其展平应该会产生符合您期望的结果。

var listRDD = sc.parallelize(finalOne)

val dataFrameForHive : DataFrame = listRDD.toDF("table_name", "constraint_applied", "data")

【讨论】：

我相信在社区 wiki 中添加它会很有帮助。