上一篇解决了 BlockManager 的存储管理。这一篇进入容错机制——Spark 如何在节点故障时恢复计算。

分布式系统的容错通常有两条路线：数据副本（每份数据存多份，一份丢了从副本恢复）和计算重放（不存副本，丢了从头重算）。Spark 选择了第二条路线——通过 RDD 的 Lineage（血统）记录计算路径，丢失的分区沿着 Lineage 重算。

本文只抓一个问题：Lineage 容错的优势和代价，以及 Checkpoint 和推测执行如何补充 Lineage 的不足。

Lineage 容错原理

每个 RDD 记录了自己的 dependencies——从哪些父 RDD 经过什么变换得来。当一个分区丢失时（Executor 故障、磁盘损坏），Spark 不需要从副本恢复，只需要找到该分区的父 RDD 分区，重新执行 compute 函数。

1
2
3
4
5
6
7
8
9
10
RDD-0 (HDFS) → RDD-1 (map) → RDD-2 (filter) → RDD-3 (reduceByKey)
                                                      │
                                                 分区 2 丢失
                                                      │
                                                      ▼
                                           沿 Lineage 回溯：
                                           RDD-3.P2 依赖 RDD-2 所有分区（宽依赖）
                                           → 如果 RDD-2 已缓存，从缓存恢复
                                           → 如果未缓存，继续回溯到 RDD-1、RDD-0
                                           → 最终从 HDFS 重读对应 Block

Lineage 容错相对数据副本的优势：不需要额外的存储空间来保存副本。对于窄依赖链条，恢复一个分区只需要重算该分区对应的父分区，不影响其他分区的计算。

但 Lineage 容错有一个结构性弱点：宽依赖。如果丢失的分区经过了 Shuffle（宽依赖），恢复该分区需要重算上游 Stage 的所有分区——因为宽依赖意味着该分区的数据来自上游所有分区。如果上游 Stage 的 Shuffle 文件也丢失了（比如对应的 Executor 也挂了），就需要从更上游重算，形成级联重算。

1
2
3
4
5
6
7
8
9
10
11
窄依赖恢复（只重算一个分区）：

  RDD-0.P2 → RDD-1.P2 → RDD-2.P2 (丢失)
  恢复路径: 只需重算 P2 这条线

宽依赖恢复（可能重算整个上游 Stage）：

  RDD-0.P0 ─┐
  RDD-0.P1 ─┤→ Shuffle → RDD-1.P2 (丢失)
  RDD-0.P2 ─┘
  恢复路径: 需要重新执行上游 Stage 所有 Task 的 Shuffle Write

Checkpoint：截断过长的 Lineage

当 Lineage 链条很长时（比如迭代算法的每轮迭代都产生新 RDD），一个分区的丢失可能触发从数据源开始的全量重算。Checkpoint 通过把 RDD 的数据写入可靠存储（通常是 HDFS）来截断 Lineage。

1
2
3
4
5
6
7
8
9
sc.setCheckpointDir("hdfs:///spark/checkpoint")

val rdd = sc.textFile("data.txt")
  .flatMap(_.split(" "))
  .map(w => (w, 1))
  .reduceByKey(_ + _)

rdd.checkpoint()  // 标记需要 checkpoint
rdd.count()       // 触发计算 + checkpoint 写入

Checkpoint 的工作流程：

1
2
3
4
5
1. rdd.checkpoint() → 打标记，不立即执行
2. action 触发计算 → 正常执行 Job
3. Job 完成后 → Spark 启动一个新 Job，把 RDD 的所有分区写入 HDFS
4. 写入完成后 → 清除该 RDD 的 dependencies，替换为 CheckpointRDD
5. 后续使用 → 从 HDFS 读取，不再沿 Lineage 重算

注意步骤 3：checkpoint 默认会触发一次额外的计算（重新执行 RDD 的计算逻辑把结果写入 HDFS）。为了避免重复计算，通常在 checkpoint 之前先 persist 到内存：

1
2
3
4
rdd.persist(StorageLevel.MEMORY_AND_DISK)
rdd.checkpoint()
rdd.count()  // 第一次计算，结果存入内存 + 写入 HDFS
             // 不会重复计算

Checkpoint 和 persist 的本质区别：persist 把数据存在 Executor 本地（内存或磁盘），Executor 故障时数据丢失；checkpoint 把数据写入 HDFS 等分布式文件系统，节点故障后仍然可以恢复。

推测执行：应对慢节点

在大规模集群中，总有一些节点因为硬件退化、网络拥塞或资源争抢而运行缓慢。一个 Stage 中如果有一个 Task 特别慢（straggler），整个 Stage 的完成时间由这个最慢的 Task 决定。

推测执行（Speculative Execution）的策略是：如果检测到某个 Task 的运行时间显著超过同 Stage 其他 Task 的中位数，在另一个 Executor 上启动该 Task 的备份副本。两个副本中先完成的被采用，后完成的被杀掉。

1
2
3
4
5
6
7
8
9
10
11
12
正常执行:
  Task 0: ████████ (2s)
  Task 1: ████████████████████████████ (7s) ← straggler
  Task 2: █████████ (2.2s)
  Stage 完成时间: 7s

推测执行:
  Task 0: ████████ (2s)
  Task 1: ████████████████████████████ (7s) ← 原始
  Task 1': ████████████ (3s) ← 备份，先完成，被采用
  Task 2: █████████ (2.2s)
  Stage 完成时间: 3s

推测执行的配置：

1
2
3
4
5
spark.speculation=true                     开启推测执行（默认 false）
spark.speculation.interval=100ms           检查间隔
spark.speculation.multiplier=3             阈值倍数（中位数的 3 倍）
spark.speculation.quantile=0.75            至少 75% 的 Task 完成后才开始推测
spark.speculation.minTaskRuntime=100ms     Task 至少运行 100ms 后才被推测

推测执行不是没有代价。备份 Task 消耗额外的 CPU 和内存资源。如果 straggler 的原因是数据倾斜（该分区数据量本身就大），备份 Task 大概率也同样慢。推测执行适合硬件异构或临时负载不均的场景，不适合数据倾斜场景。

Shuffle 文件的容错

Shuffle 文件存储在 Executor 的本地磁盘上。如果 Executor 故障，它的 Shuffle 文件也随之丢失。下游 Stage 的 Task 在读取 Shuffle 数据时会发现 FetchFailedException，DAGScheduler 会重新提交丢失了 Shuffle 输出的上游 Stage。

External Shuffle Service（ESS）可以缓解这个问题：把 Shuffle 文件的管理从 Executor 进程分离到一个独立的服务进程中。即使 Executor 被 kill（比如动态资源分配回收空闲 Executor），Shuffle 文件仍然可以通过 ESS 读取。

1
2
3
4
5
无 ESS:
  Executor 故障 → Shuffle 文件丢失 → 上游 Stage 重算

有 ESS:
  Executor 故障 → Shuffle 文件仍在 ESS 管理下 → 下游正常读取

ESS 在启用动态资源分配（Dynamic Resource Allocation）时几乎是必需的——否则每次释放 Executor 都会丢失 Shuffle 数据。

实验：观察容错行为

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
// 观察 checkpoint 截断 Lineage
sc.setCheckpointDir("/tmp/spark-checkpoint")

val rdd = sc.parallelize(1 to 100, 4)
  .map(_ * 2)
  .filter(_ > 50)
  .map(_.toString)

println("Before checkpoint:")
println(rdd.toDebugString)
// 完整的 Lineage 链

rdd.checkpoint()
rdd.count()

println("\nAfter checkpoint:")
println(rdd.toDebugString)
// Lineage 被截断，根节点变成 ReliableCheckpointRDD

checkpoint 前，toDebugString 显示从 parallelize 开始的完整变换链。checkpoint 后，Lineage 被截断为 ReliableCheckpointRDD → 当前 RDD，之前的所有父 RDD 信息都被丢弃。

模式提炼

Spark 的 Lineage 容错体现了计算重放 vs 数据副本的设计选择。

计算重放的优势是不需要额外的存储开销来维护副本，在窄依赖场景下恢复效率高。代价是宽依赖场景下可能触发级联重算。Checkpoint 通过在关键节点物化数据来平衡这个代价。

这个选择和数据库的 WAL（Write-Ahead Log）恢复机制形成对比：数据库先写日志，再更新数据页。崩溃后从日志重放（redo）恢复数据。Spark 的 Lineage 相当于一种"声明式的日志"——不记录具体操作的序列，而是记录数据之间的变换关系。恢复时不是"重放操作"，而是"重算数据"。

推测执行体现了冗余计算换确定性延迟的通用策略。这个思路在分布式系统中广泛使用：Google 的 “The Tail at Scale” 论文详细讨论了如何用冗余请求对抗长尾延迟。

工程迁移表

Spark 概念 MapReduce 对应 Flink 对应 数据库对应
Lineage 容错 Task 重试（从 HDFS 输入重算） Checkpoint + 状态恢复 WAL Redo 日志
Checkpoint 无（每轮结果写 HDFS） 周期性 Checkpoint（Chandy-Lamport） 检查点（数据库 Checkpoint）
推测执行 Speculative Execution（同名机制） 无内置推测执行 查询超时重试
External Shuffle Service 无（MapReduce 不分离 Shuffle） Remote Shuffle Service 独立的中间结果存储
级联重算 Job 间通过 HDFS 隔离，不级联 从最近 Checkpoint 恢复，不重算 从最近检查点开始 Redo

常见误解

误解一：“Lineage 等价于操作日志”。Lineage 记录的是 RDD 之间的变换关系（声明式），不是具体的操作序列（命令式）。它不记录"读了哪些字节、写了哪些字节"，只记录"从哪个 RDD 经过什么函数得到当前 RDD"。

误解二：“Checkpoint 和 persist(DISK_ONLY) 效果相同”。persist(DISK_ONLY) 把数据写入 Executor 本地磁盘，Executor 故障时数据丢失。Checkpoint 写入 HDFS，节点故障后仍然可用。更关键的区别是 checkpoint 会截断 Lineage，而 persist 不会。

误解三：“开启推测执行总是有益的”。在数据倾斜场景下，推测执行的备份 Task 处理的数据量和原始 Task 相同，不会更快。反而浪费了一份 Executor 资源。推测执行适合硬件性能不均匀导致的 straggler，不适合数据分布不均匀导致的 straggler。

练习

  1. 构造一条包含 5 次 map 的长 Lineage 链，用 toDebugString 观察。然后在中间某个 RDD 上 checkpoint，再次观察 toDebugString 的变化。验证 Lineage 确实被截断。

  2. 启动一个多 Executor 的 Spark 应用，缓存一个 RDD 后执行 action。在 Spark UI 的 Executors 页面手动 kill 一个 Executor（如果允许），观察后续 action 是否触发了 Task 重算。在 Stages 页面查看是否出现了 “skipped” 的 Task（表示使用了缓存）和重新执行的 Task。

  3. 思考：如果一个 iterative 算法执行 100 轮迭代，每轮产生一个新 RDD。不做 checkpoint 时，第 100 轮丢失一个分区需要从第 1 轮开始重算。如果每 10 轮 checkpoint 一次，最坏情况下需要从哪一轮开始重算？

系列导航

序号 主题 状态
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：两种计算引擎的设计选择

参考资料