2、DataFrame 基础：概念、创建方式与 Schema 定义

好，咱们今天聊聊 DataFrame。说实话，这是 Spark SQL 里最核心的概念之一。我刚开始接触 Spark 时，总觉得 DataFrame 不就是个表格嘛，有啥特别的？后来踩过坑才明白——它远不止一张表那么简单。

2.1 DataFrame 到底是什么？

说白了，DataFrame 就是一个分布式的、带有 Schema 信息的不可变数据集。你想想看，它像不像数据库里的一张表？有行有列，每列有名字、有类型。但区别在于——它跑在集群上，数据可能分散在几十台机器里。

我个人习惯把 DataFrame 理解成「带元数据的 RDD」。RDD 你知道吧？它只管数据，不管结构。而 DataFrame 在 RDD 的基础上，加上了 Schema——也就是每列叫什么、是什么类型。这个小小的改进，带来的性能提升是巨大的。

核心要点：DataFrame = RDD + Schema + 优化器（Catalyst）

为什么说它比 RDD 快？因为有了 Schema，Spark 就能做列式存储和谓词下推。我在项目中遇到过这样一个场景：用 RDD 处理 10TB 的日志数据，跑了 3 小时没跑完。换成 DataFrame 后，同样的逻辑只用了 40 分钟。嗯，这就是 Schema 带来的威力。

2.2 创建 DataFrame 的几种方式

创建 DataFrame 的方式很多，我挑最常用的三种来讲。你工作中 90% 的场景，用这三种就够了。

2.2.1 从 RDD 转换

这是最基础的方式。你手头有 RDD，想把它变成 DataFrame，怎么做？

// 方式一：通过反射推断 Schema（需要 case class）
case class Person(name: String, age: Int)
val rdd = sc.textFile("people.txt")
           .map(_.split(","))
           .map(p => Person(p(0), p(1).trim.toInt))
val df = rdd.toDF()

// 方式二：通过编程方式指定 Schema
import org.apache.spark.sql.types._
val schema = StructType(Array(
  StructField("name", StringType, nullable = false),
  StructField("age", IntegerType, nullable = true)
))
val rowRDD = rdd.map(p => Row(p(0), p(1).trim.toInt))
val df = spark.createDataFrame(rowRDD, schema)

我个人更推荐第二种方式。为什么？因为 case class 有 22 个字段的限制，而且反射推断有时会猜错类型。我曾经在生产环境里遇到过——Spark 把时间戳字段推断成了 String，导致后续所有时间计算都报错。从那以后，我但凡涉及生产任务，都老老实实手写 Schema。

小技巧：如果你不确定字段类型，可以用 df.printSchema() 看一眼。这招排查问题特别好用。

2.2.2 从数据源读取

这才是实际工作中最常用的方式。Spark SQL 支持的数据源太多了——CSV、JSON、Parquet、ORC、JDBC……我一个个说。

读取 CSV：

val df = spark.read
  .option("header", "true")
  .option("inferSchema", "true")
  .option("delimiter", ",")
  .csv("hdfs://path/to/file.csv")

这里有个坑——inferSchema 虽然方便，但会额外扫描一遍数据。如果你的文件很大（比如几百 GB），这个开销可不小。我建议小文件用 inferSchema，大文件还是手动指定 Schema 更靠谱。

读取 JSON：

val df = spark.read
  .option("multiLine", "true")  // 如果 JSON 是多行的
  .json("hdfs://path/to/data.json")

读取 Parquet（我最爱的格式）：

val df = spark.read.parquet("hdfs://path/to/data.parquet")

Parquet 是列式存储，压缩率高，读取快。我在做数仓项目时，所有中间结果都存成 Parquet。你想想看，同样的数据，CSV 存 100GB，Parquet 可能只要 20GB，而且查询速度还快 3-5 倍。何乐而不为？

数据源 推荐场景 注意事项
CSV 日志、导出数据 注意分隔符、引号转义
JSON API 输出、嵌套数据 大文件建议用 multiLine
Parquet 数仓、中间结果 自带 Schema，无需指定
JDBC 从数据库读取 注意分区键，避免全表扫描

2.2.3 从其他 DataFrame 转换

这个其实不算「创建」，但工作中很常见。比如你有一个 DataFrame，想取其中几列生成新的 DataFrame：

val df1 = df.select("name", "age")
val df2 = df.filter($"age" > 18)
val df3 = df.groupBy("city").agg(count("*"))

记住一点：DataFrame 是不可变的。每次操作都生成新的 DataFrame，原来的不会变。这个特性一开始可能不习惯，但习惯了就觉得——嗯，这样反而安全，不会莫名其妙把原始数据改坏了。

2.3 Schema 的定义与操作

Schema 就是 DataFrame 的「骨架」。没有 Schema，Spark 就不知道每列是什么类型，也就没法做优化。

2.3.1 手动定义 Schema

前面提到过，我推荐手动定义 Schema。来看看怎么写：

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

val schema = StructType(Array(
  StructField("id", LongType, nullable = false),
  StructField("name", StringType, nullable = true),
  StructField("age", IntegerType, nullable = true),
  StructField("salary", DoubleType, nullable = false),
  StructField("hire_date", DateType, nullable = true)
))

这里每个 StructField 有三个参数：字段名、字段类型、是否允许为空。我建议你养成习惯——能不为空就不为空。为什么？因为 nullable = true 会让 Spark 做额外的空值检查，影响性能。而且从业务角度讲，关键字段（比如 ID、金额）本来就不该为空。

注意：Spark 的 DateType 和 TimestampType 容易搞混。DateType 只存日期（yyyy-MM-dd），TimestampType 存日期+时间。我曾经因为用错类型，导致时间窗口计算差了 8 小时——因为时区问题。嗯，这个坑我帮你踩过了。

2.3.2 查看和修改 Schema

查看 Schema 很简单：

df.printSchema()
// 输出：
// root
//  |-- id: long (nullable = false)
//  |-- name: string (nullable = true)
//  |-- age: integer (nullable = true)

// 获取 Schema 对象
val schema = df.schema

修改 Schema 呢？比如你想改列名、改类型：

// 改列名
val df2 = df.withColumnRenamed("name", "user_name")

// 改类型
val df3 = df.withColumn("age", $"age".cast(IntegerType))

// 新增列
val df4 = df.withColumn("tax", $"salary" * 0.1)

这里有个细节——withColumn 每次都会生成新的 DataFrame。如果你要加 10 列，就会有 10 次 Shuffle？其实不会。Catalyst 优化器会把这些操作合并成一个执行计划。所以放心用，别担心性能。

2.3.3 Schema 的合并与演化

这个知识点比较进阶，但很重要。当你读取多个 Parquet 文件时，它们的 Schema 可能不一样——比如旧文件有 5 列，新文件有 6 列。怎么办？

val df = spark.read
  .option("mergeSchema", "true")
  .parquet("hdfs://path/to/data/")

设置 mergeSchema = true，Spark 会自动合并所有文件的 Schema。缺失的列用 null 填充。这个功能我在做增量数据导入时经常用——每天的数据可能新增字段，用 mergeSchema 就不用改代码。

不过要注意：mergeSchema 有性能开销。它会扫描所有文件的元数据。如果文件数量特别多（比如上万个），建议还是统一 Schema 比较好。

2.4 小结

好了，关于 DataFrame 基础，我总结几个要点：

  • DataFrame 是带 Schema 的分布式数据集，比 RDD 更高效、更易用
  • 创建方式有三种：从 RDD 转换、从数据源读取、从其他 DataFrame 衍生
  • Schema 是 DataFrame 的灵魂，建议手动定义，不要依赖 inferSchema
  • Parquet 是首选存储格式，性能好、自带 Schema

下一章我们会深入 DataFrame 的常用操作——select、filter、groupBy 这些。到时候我会分享一些实战中的优化技巧。咱们下期见。

← 上一章 📖 返回目录 下一章 →