Scala中如何将case class对象数组转换为DataFrame

Spark Scala API原生支持直接将存储case class的序列类型转换为DataFrame，case class的字段会自动映射为DataFrame的schema，不需要手动定义列类型。

最短实现代码 #

// 假设你已经创建好SparkSession实例命名为spark
import spark.implicits._

// 你的ArrayBuffer实例
val rowBuffer: ArrayBuffer[Rows] = ??? // 你已经填充好数据的集合
val df = rowBuffer.toDF()

完整可运行示例 #

import org.apache.spark.sql.SparkSession
import scala.collection.mutable.ArrayBuffer

// 样例类建议定义在方法外的顶层作用域，避免序列化问题
case class Rows(column: String, operation: String, result: String)

object Demo {
  def main(args: Array[String]): Unit = {
    // 初始化SparkSession
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("buffer-to-df")
      .getOrCreate()

    // 必须导入SparkSession的隐式转换，否则无法识别toDF方法
    import spark.implicits._

    // 模拟填充数据的ArrayBuffer
    val rowBuffer = ArrayBuffer(
      Rows("user_id", "count", "12000"),
      Rows("pay_amount", "sum", "3400000"),
      Rows("refund_rate", "avg", "0.023")
    )

    // 转换得到DataFrame
    val resultDf = rowBuffer.toDF()

    // 验证输出
    resultDf.printSchema()
    resultDf.show(false)

    spark.stop()
  }
}

常见问题说明 #

• 必须在创建SparkSession实例之后导入spark.implicits._，提前导入或者导入错包会导致toDF()方法编译报错
• 样例类Rows不要定义在main方法等局部作用域，部分低版本Spark无法正确解析局部定义的case class结构，还会触发序列化错误
• 如果字段可能存在空值，将对应字段类型声明为Option[T]即可，比如result: Option[String]，Spark会自动将None转换为DataFrame的null值
• 如果隐式转换因为环境问题无法生效，可以用手动创建的API替代：

  val resultDf = spark.createDataFrame(rowBuffer, classOf[Rows])
  

内容的提问来源于stack exchange，提问作者Shuvam Pandey