61.2 Spark 计算模型
一句话
Spark 的核心只有两条：RDD 血统（lineage）记录 “怎么算出来”、DAG 按宽依赖切 stage；同一个需求，性能差异几乎全部来自 “shuffle 了多少字节”——写法只差一行，代价可能差 25 倍。
窄依赖与宽依赖
| 依赖 | 含义 | 是否 shuffle | 是否切 stage | 例子 |
|---|---|---|---|---|
| 窄依赖 | 每个父分区最多被一个子分区使用 | 否 | 否 | map、filter、union |
| 宽依赖 | 子分区依赖多个父分区 | 是 | 是 | groupByKey、join（未广播）、repartition |
stage 的边界就是宽依赖：一个 stage 内部可以流水线执行（不落盘），跨 stage 必须 shuffle（落盘 + 网络传输）。
实测：一行写法差 25 倍 shuffle
5000 万条记录、每条 100B：
| 写法 | shuffle 字节 | stage 数 |
|---|---|---|
map / filter（窄依赖） | 0 | 1 |
groupByKey | 5.0GB | 2 |
reduceByKey（map 端预聚合） | 0.2GB | 2 |
reduceByKey 与 groupByKey 语义相近，但前者在 map 端先做局部聚合，shuffle 只剩 “每个 key 一个值” 而不是 “每个 key 的所有值”。25 倍的差距就是这么来的。
// ❌ 全量 shuffle：每个 key 的所有 value 都要过网络
rdd.groupByKey().mapValues(_.sum)
// ✅ map 端预聚合：先局部求和，再 shuffle
rdd.reduceByKey(_ + _)
三个决定性能的选择
| 选择 | 默认 | 何时改 |
|---|---|---|
| 分区数 | 由父 RDD 决定 | shuffle 后分区太少 → 单 task 数据量大；太多 → 小文件与调度开销 |
| join 策略 | 大表 join 大表 = shuffle join | 小表 → broadcast join（不发 shuffle） |
| 缓存 | 不缓存 | 同一 RDD 被多次使用 → persist()（内存/磁盘） |
反模式
| 反模式 | 后果 |
|---|---|
groupByKey 做聚合 | shuffle 全量数据，网络与磁盘爆炸 |
collect() 拉全量结果到驱动 | OOM，驱动成瓶颈 |
| 小表 join 大表不用 broadcast | 白做一次大 shuffle |
频繁 repartition | 每次都全量 shuffle |
| 缓存所有中间结果 | 内存被占满，反而频繁 spill |
| 分区数固定写死 | 数据量变化后 task 过大/过小 |
三条铁律
🔑 性能问题先看 shuffle 字节，再看 CPU。
🔑 能在 map 端聚合的，绝不留到 reduce 端。
🔑 小表广播、大表分区；join 策略比 “调参” 影响大。
自检
Q1 为什么 reduceByKey 比 groupByKey 快这么多？
▸ 答案：因为 reduceByKey 在 map 端先做局部聚合（combine）：每个分区先算出 “每个 key 的局部和”，shuffle 只传 “key + 局部和”；groupByKey 则把 “key 的所有 value” 原样传过去。实测差距 5.0GB → 0.2GB，约 25 倍。
Q2 为什么 “宽依赖” 必须切 stage？
▸ 答案：因为宽依赖需要按 key 重新分布数据：下游分区要拿到所有上游分区中对应 key 的数据，必须等 shuffle 完成。stage 边界就是 “必须落盘/等网络” 的地方，也是失败重算的边界。
Q3 分区数怎么定？
▸ 答案：按 “每个 task 处理的数据量” 定：经验值是每分区 100–200MB（可配），太小则调度开销大、小文件多；太大则单 task 慢、易 OOM。另外要考虑核数：分区数最好是 “总核数的 2–4 倍”，让 task 能并行填满。
Q4 broadcast join 为什么能省掉 shuffle？
▸ 答案：因为它把小表完整复制到每个 executor，大表的每个分区在本地与小表做 join——不需要按 key 重分布大表，所以没有 shuffle。前提是小表能放进内存（通常几十 MB 到几百 MB），否则广播本身会成为瓶颈。
Q5 cache() 与 persist() 的区别，什么时候必须用？
▸ 答案：cache() 等价于 persist(MEMORY_AND_DISK)（版本略有差异）；persist() 可选存储级别。必须用的场景是 “同一个 RDD 被多个 action 复用”（如反复迭代的机器学习、多次查询的中间结果），否则每次 action 都会从头重算血统。
Q6 Spark 为什么比 MapReduce 快，又不总是快？
▸ 答案：快在 DAG + 内存流水线：窄依赖不落盘、多 stage 一次调度；MapReduce 每个 job 都要落盘。但 “不总是快” 是因为：shuffle 仍要落盘、内存不足会 spill、JVM 与 GC 开销、以及 “小作业启动开销” 相对更高。
Q7 血统（lineage）丢了会怎样？
▸ 答案：血统是容错机制：分区丢失时按血统重算。如果链路太长（几百个转换），重算代价极大，所以要 checkpoint() 截断血统；如果血统被截断又没有 checkpoint，丢失分区就无法恢复，只能重跑整个作业。