上一篇解决了 Catalyst 优化器的四阶段流水线。这一篇进入用户 API 层——RDD、DataFrame、Dataset 三代 API 的演进逻辑。

三者的区别容易被简化为"RDD 是低级 API，DataFrame 是高级 API"。更准确的说法是：三代 API 在类型安全和执行优化之间做了不同的取舍。RDD 完全类型安全但无法被 Catalyst 优化；DataFrame 完全被 Catalyst 优化但放弃了编译期类型检查；Dataset 试图兼顾两者，但付出了 Encoder 序列化/反序列化的代价。

本文只抓一个问题：三代 API 在内部表示、优化路径和序列化机制上的具体差异。

三代 API 的内部表示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
RDD[Person]
  └─ 内部存储: JVM 堆上的 Java/Scala 对象
  └─ 优化路径: 无（用户代码是黑盒，Spark 无法查看函数内部）
  └─ 类型信息: 编译期完整保留（泛型参数 T = Person）

DataFrame (= Dataset[Row])
  └─ 内部存储: UnsafeRow（Tungsten 二进制格式）
  └─ 优化路径: 完整经过 Catalyst 四阶段
  └─ 类型信息: 运行时 schema（StructType），编译期只知道是 Row

Dataset[Person]
  └─ 内部存储: UnsafeRow + Encoder（需要时反序列化为 Person 对象）
  └─ 优化路径: 关系操作经过 Catalyst，lambda 函数仍是黑盒
  └─ 类型信息: 编译期保留（泛型参数 T = Person）

关键区别在于"黑盒"问题。当用户传入一个 lambda 函数（如 dataset.map(person => person.age + 1)），Catalyst 无法解析 lambda 内部的逻辑。它只知道"这是一个从 Person 到 Int 的函数"，不知道函数访问了哪个字段、做了什么运算。因此无法做谓词下推或列裁剪。

相比之下，DataFrame 的列操作（如 df.select($"age" + 1)）使用 Spark 的 Expression 对象构建，Catalyst 可以完全解析和优化。

DataFrame：无类型但可优化

DataFrame 是 Spark 1.3 引入的 API，本质是 Dataset[Row] 的类型别名。Row 是一个无类型的容器，按索引或列名访问字段。

1
2
3
4
5
6
7
val df = spark.read.parquet("users.parquet")
// df 的类型是 DataFrame (= Dataset[Row])
// 编译器只知道它包含 Row，不知道有哪些列

val result = df.filter($"age" > 18).select($"name", $"age")
// $"age" > 18 构建的是一个 GreaterThan(AttributeReference("age"), Literal(18))
// Catalyst 可以完整解析这个表达式

DataFrame 的劣势是类型错误延迟到运行时。如果写了 df.select($"agee")（拼写错误），编译器不会报错，只有运行时 Analysis 阶段才会抛出 AnalysisException。

Dataset：类型安全但有序列化代价

Dataset 是 Spark 1.6 引入的 API，试图在 DataFrame 的优化能力和 RDD 的类型安全之间找平衡。

1
2
3
4
5
6
7
8
9
case class Person(name: String, age: Int)
val ds: Dataset[Person] = spark.read.parquet("users.parquet").as[Person]

// 类型安全的操作——编译期检查
val adults = ds.filter(_.age > 18)  // lambda: Person => Boolean
val names = ds.map(_.name)          // lambda: Person => String

// 关系操作——经过 Catalyst 优化
val result = ds.filter($"age" > 18).select($"name")

Dataset 的核心机制是 Encoder。Encoder 知道如何在 UnsafeRow（二进制格式）和 JVM 对象（如 Person）之间转换。当用户调用 lambda 操作（如 map、filter with lambda）时，Spark 需要把 UnsafeRow 反序列化为 Person 对象，调用 lambda，再把结果序列化回 UnsafeRow。

1
2
3
4
5
6
Dataset[Person] 的 lambda 操作数据流：

  UnsafeRow → Encoder.fromRow() → Person 对象 → lambda(person) → 结果
      ↑                                                            │
      │                                                            ▼
  Tungsten 二进制                                        Encoder.toRow() → UnsafeRow

这个反序列化-序列化的往返是 Dataset lambda 操作的性能代价。如果整个查询都用关系操作（select、filter with column expression、groupBy、agg），Dataset 和 DataFrame 的性能完全相同，因为不触发 Encoder 转换。

实验：对比三种 API 的执行计划

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
case class WordCount(word: String, count: Int)

// 准备数据
val data = Seq(("hello", 1), ("world", 2), ("hello", 3))

// RDD API
val rdd = sc.parallelize(data)
val rddResult = rdd.reduceByKey(_ + _).collect()

// DataFrame API
val df = data.toDF("word", "count")
df.groupBy("word").agg(sum("count")).explain(true)
// Physical Plan 中可以看到 HashAggregate with Catalyst 优化

// Dataset API (关系操作)
val ds = data.toDF("word", "count").as[WordCount]
ds.groupByKey(_.word).reduceGroups((a, b) => WordCount(a.word, a.count + b.count))
  .explain(true)
// Physical Plan 中可以看到 DeserializeToObject 节点——Encoder 开销的标志

在 explain 输出中，如果看到 DeserializeToObject 和 SerializeFromObject 节点，说明发生了 Encoder 转换。纯关系操作（select/filter with column）的计划中不会出现这些节点。

选择指南

1
2
3
4
5
6
7
场景                           推荐 API        原因
────────────────────────────── ─────────────── ───────────────────────────
ETL / SQL 查询 / BI 报表        DataFrame       全路径 Catalyst 优化，无类型开销
结构化数据处理 + 类型安全需求    Dataset         编译期检查，关系操作时性能等同 DataFrame
需要 lambda 的复杂业务逻辑      Dataset         类型安全 + 可读性，接受 Encoder 代价
非结构化数据 / 自定义分区逻辑    RDD             Catalyst 无法优化的场景
交互式数据探索                  DataFrame       schema 动态发现，列名即可操作

一条务实的建议：默认用 DataFrame（关系操作），需要类型安全时用 Dataset（关系操作优先、lambda 备选），只有 Catalyst 真的无法表达的逻辑才退回 RDD。

模式提炼

三代 API 的演进体现了一个通用的设计张力：抽象级别 vs 优化空间。

抽象级别越高（RDD 的任意 lambda），系统越不了解用户意图，优化空间越小。抽象级别越受限（DataFrame 的关系操作），系统对用户意图的理解越完整，优化空间越大。Dataset 尝试用 Encoder 机制在两者之间架桥，但桥的两端仍然有转换代价。

同样的张力在其他系统中反复出现：SQL vs 存储过程（SQL 可优化但表达力受限，存储过程灵活但是优化器黑盒）、React 的声明式 JSX vs 命令式 DOM 操作、TensorFlow 的静态图 vs PyTorch 的动态图。

工程迁移表

Spark 概念 Flink 对应 数据库对应
RDD[T]（类型安全、无优化） DataStream（处理函数是黑盒） 存储过程（优化器无法介入）
DataFrame（无类型、全优化） Table API / Flink SQL SQL 查询（全路径优化）
Dataset[T]（折中） 无直接对应 带类型映射的 ORM 查询
Encoder Flink 的 TypeSerializer JDBC ResultSet 映射
UnsafeRow Flink 的 BinaryRow 数据库的行存储格式

常见误解

误解一：“Dataset 总是比 DataFrame 慢”。只有使用 lambda 操作时 Dataset 才有额外的 Encoder 开销。使用关系操作（select、filter with column、groupBy.agg）时，Dataset 和 DataFrame 走完全相同的执行路径，性能无差异。

误解二：“RDD API 已经废弃”。RDD API 没有废弃，但不再是推荐的主要 API。Spark 内部仍然在 RDD 层面执行物理计划——DataFrame/Dataset 最终被翻译成 RDD 操作。某些无法用关系操作表达的逻辑（如自定义分区器、非结构化数据处理）仍然需要 RDD。

误解三：“DataFrame 和 Dataset 可以随意转换而无代价”。DataFrame 到 Dataset（.as[T]）的转换是懒惰的，本身没有代价。但转换后如果使用 lambda 操作，就会触发 Encoder 的反序列化。从 Dataset 到 DataFrame（.toDF()）的转换也是懒惰的，没有实际的数据转换。

练习

  1. 对同一份百万行数据，分别用 RDD 的 map + reduceByKey、DataFrame 的 groupBy.agg、Dataset 的 groupByKey.reduceGroups 做 wordcount。在 Spark UI 中对比三者的执行时间和 Stage 数量。

  2. 对一个 Dataset[Person]，分别用 ds.filter(_.age > 18)（lambda）和 ds.filter($"age" > 18)（column expression）执行过滤。用 explain(true) 对比两者的物理计划，找出 DeserializeToObject 节点的有无。

  3. 思考：如果 Spark 能在编译期自动把 lambda _.age > 18 解析为 column expression $"age" > 18，Dataset 的 Encoder 代价是否可以完全消除？Scala 3 的 macro 或 inline 机制是否可以实现这一点？

系列导航

序号 主题 状态
00 导读：为什么 Spark 的核心是一张 DAG 已完成
01 RDD：弹性分布式数据集的五大属性 已完成
02 宽依赖与窄依赖：Stage 是怎样划分出来的 已完成
03 DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行 已完成
04 Shuffle 机制：数据跨分区交换的代价与优化 已完成
05 内存管理与 Tungsten：堆外内存、序列化与代码生成 已完成
06 存储体系：Block Manager、广播变量与累加器 已完成
07 容错机制：Lineage、Checkpoint 与推测执行 已完成
08 Spark SQL 与 Catalyst：从 SQL 文本到物理算子树 已完成
09 DataFrame 与 Dataset：类型安全与性能的折中 本篇
10 Adaptive Query Execution：运行时改写执行计划 下一篇
11 数据源 API 与谓词下推：让存储层帮忙过滤
12 Structured Streaming：把流当成无界表
13 水位线与状态管理：流处理的时间语义
14 Spark on YARN / K8s：资源管理与部署模式
15 动态资源分配与作业调度：多租户集群的资源博弈
16 性能诊断：Spark UI、Event Log 与调优思路
17 Spark 与 Flink：两种计算引擎的设计选择

参考资料