Spark Scala中RDD转换为DataFrame的正确实现方法

sc.textFile() 读取生成的是RDD[String]，每个元素为整行文本字符串，既没有按逗号拆分业务字段，也没有定义字段的名称、类型约束，直接调用toDF()只会生成仅含单个value字符串列的DataFrame，完全不符合4字段的业务结构，必然无法得到预期结果。

Scala环境下RDD转DataFrame有两种通用成熟方案，根据场景选择即可： #

方案1：Case Class反射生成Schema（固定字段场景首选，代码最简洁） #

这个方案依赖Spark的隐式转换自动映射结构，注意不要漏掉隐式导入：

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

// 定义与数据字段一一对应的样例类，字段类型按业务需求指定
case class Customer(
  custId: Int,
  custName: String,
  custEmail: String,
  custPhone: String
)

val customerDF = sc.textFile("file:///home/hduser/data/customer.txt")
  .map(line => {
    val parts = line.split(",")
    // 逐行拆分字段，转为Customer类型对象
    Customer(
      parts(0).trim.toInt,
      parts(1).trim,
      parts(2).trim,
      parts(3).trim
    )
  })
  .toDF()

// 验证转换结果
customerDF.printSchema()
customerDF.show()

样例类的字段名会直接作为DataFrame的列名，Scala基础类型会自动映射为Spark SQL对应的数据类型，日常开发固定结构的场景优先用这个方案。 #

方案2：编程式定义Schema（动态字段场景使用） #

如果字段结构是动态生成、无法提前定义样例类，可以手动构造Schema调用createDataFrame方法：

import org.apache.spark.sql.Row
import org.apache.spark.sql.types._

// 先逐行拆分数据，转为RDD[Row]类型
val rowRDD = sc.textFile("file:///home/hduser/data/customer.txt")
  .map(line => {
    val parts = line.split(",")
    Row(
      parts(0).trim.toInt,
      parts(1).trim,
      parts(2).trim,
      parts(3).trim
    )
  })

// 手动构造Schema，字段顺序必须和Row中元素的顺序严格一致
val schema = StructType(Seq(
  StructField("custId", IntegerType, nullable = false),
  StructField("custName", StringType, nullable = true),
  StructField("custEmail", StringType, nullable = true),
  StructField("custPhone", StringType, nullable = true)
))

val customerDF = spark.createDataFrame(rowRDD, schema)

• 导入隐式转换时，spark要替换为你自己代码中初始化的SparkSession实例名。
• 原代码路径中data//customer.txt存在重复斜杠，建议修正为单斜杠避免路径解析异常。
• 生产环境使用时建议增加字段长度校验，过滤字段数不足4个的脏数据，避免数组下标越界报错。
• 手机号字段如果存在非数字字符、或者长度超过Int范围，不要用数值类型存储，保持String类型即可。

内容的提问来源于stack exchange，提问作者Dhrumil Shah