【Spark Core 笔记】RDD 原理、调度流程和线上踩坑

这篇笔记整理了用 Spark 的一些心得，覆盖了面试常考的核心知识点，也包括了不少线上真实踩过的坑。

一、Spark Core 到底是什么

先给一个我自己的理解：Spark Core 是整个 Spark 生态的"发动机"。不管是后面你用的 Spark SQL、Spark Streaming 还是 MLlib，底层的执行引擎都是 Core 这一套东西。它解决的核心问题就两个：

1. 如何把一个大计算任务拆成小块，分布到多台机器上并行跑
2. 跑的过程中某个节点挂了，怎么自动恢复而不是从头再来

这两个问题分别对应分布式计算和容错机制，也是 Spark 区别于 MapReduce 最核心的竞争力。

二、RDD —— 一切计算的起点

2.1 RDD 的本质

RDD（Resilient Distributed Dataset，弹性分布式数据集），名字很长，但其实理解起来不复杂。你可以把它想象成一个被切成了很多份的不可变数据集合，每一份存到不同的节点上，你对它的每一次操作都会生成一个新的 RDD。

这里有几个关键词需要注意：

2.2 RDD vs DataFrame：不是谁更好，是不同阶段的选择

很多新手会纠结：既然 DataFrame 有 Catalyst 优化器，那还用 RDD 干嘛？

一个实用的判断标准：如果你的数据已经是结构化的（JSON、Parquet、Hive 表），直接上 DataFrame/Dataset；如果你需要精确控制数据的分布、分区策略，或者做一些非标准的数据处理（比如图计算），RDD 仍然是最灵活的选择。

2.3 宽窄依赖 —— Stage 划分的核心依据

这是面试必考题，也是理解 Spark 执行流程的关键。

窄依赖（Narrow Dependency）：父 RDD 的每个 partition 最多被子 RDD 的一个 partition 使用。典型代表：map、filter、union、mapPartitions。

父 RDD: [A][B][C] 子 RDD: [A'][B'][C'] | | +--一对一映射------+

窄依赖的特点是可以在同一个 Stage 内流水线执行，不需要等待父 RDD 全部算完，父 RDD 一个 partition 算完直接传给子 RDD 的算子继续处理。

宽依赖（Wide Dependency）：父 RDD 的 partition 会被子 RDD 的多个 partition 使用。典型代表：groupByKey、reduceByKey、sortByKey、join（非 co-partitioned 情况下）。

父 RDD: [A][B] 子 RDD: [结果1][结果2][结果3][结果4] | \ / | | | \ / | | +----+Shuffle----+----+----+

宽依赖的特点是必须等父 RDD 所有 partition 的数据都准备好，经过 Shuffle 才能继续，所以它是 Stage 的边界线。

2.4 DAG 怎么切 Stage？Task 怎么生成？

DAGScheduler 切 Stage 的规则：从最后一个 RDD 往前推，遇到宽依赖就切一刀。

RDD A -> map -> RDD B -> map -> RDD C -> groupByKey -> RDD D -> map -> RDD E -> collect() ^宽依赖切一刀^ ^Action触发^ Stage 0: A.map().map() (窄依赖流水线) Stage 1: C.groupByKey() (Shuffle) Stage 2: D.map() -> E (窄依赖流水线)

Task 的生成：每个 Stage 内，根据 RDD 的 partition 数量生成对应数量的 Task。比如 Stage 0 有 100 个 partition，就会生成 100 个 Task 分发到 Executor 上并行执行。

三、Transformation 和 Action —— 懒执行的秘密

3.1 核心区别

3.2 懒执行怎么实现的？

Spark 内部维护了一个DAG（有向无环图）。每次调用 Transformation，并不会真正跑计算，而是往 DAG 里添加一个节点，记录"这个 RDD 是从哪个父 RDD 通过什么算子变来的"。只有当碰到 Action 时，Spark 才会把这个 DAG 从头到尾扫描一遍，优化后切成 Stage，提交到集群执行。

这个设计的好处：Spark 有机会看到完整的计算图，做全局优化。比如你把filter和map连续调用，Spark 可以合并成一个 Stage 流水线执行，而不是真的写两遍中间结果。

3.3 常见的坑

坑 1：在 Transformation 里做外部 IO

// 错误示范：Transformation 不会立即执行，你的日志可能打不出来rdd.map{record=>println("处理记录: "+record)// 在 Executor 端执行，但你可能在日志里看不到writeToExternalDB(record)// 这也是懒执行的，不会立刻写}

正确做法：如果需要确保副作用执行，包在 Action 里或者配合foreachPartition。

坑 2：连续调 Action 导致重复计算

valrdd=source.map(...).filter(...).groupByKey()valcount=rdd.count()// Action 1：触发完整计算valresult=rdd.collect()// Action 2：又触发一遍完整计算！

两次 Action 之间 RDD 的血统没有缓存，所以count()和collect()都会从头算一遍。解决方式是在中间加rdd.cache()或rdd.persist()。

四、任务调度全流程 —— 从代码到集群执行

4.1 先搞清楚 5 个角色各自干啥

4.2 完整执行流程（9 步，跟着走一遍就懂了）

Step 1: 用户 spark-submit 提交 Application → 比如: spark-submit --class MyApp --master yarn myapp.jar Step 2: Driver 启动（主控进程，通常跑在集群某台机器的 Container 里） → 解析用户代码，构建 DAG 逻辑执行计划 → 维护 RDD 之间的依赖关系（Lineage） → 注意: 此时只是解析，还没有真正执行！ Step 3: Driver 向 Cluster Manager 申请资源 → CM 可以是 YARN / Kubernetes / Standalone → 请求在各个 Worker 节点上启动 Executor 进程 Step 4: Cluster Manager 在 Worker 节点启动 Executor → Executor（工作节点）启动后向 Driver 注册 → Executor 内部有线程池 + BlockManager（存 cache/shuffle 数据） → 多个 Worker 节点同时启动，形成分布式执行集群 Step 5: 代码中遇到第一个 Action，触发真正计算 → Action 比如: collect() / count() / saveAsTextFile() → 之前所有 Transformation 都是懒执行的，只有 Action 才触发计算 → 一个 Action = 一个 Job Step 6: DAGScheduler 开始切分 Stage → 从最后一个 RDD 往前推，遍历依赖链 → 遇到宽依赖（Shuffle）就切一刀 → 每个 Stage 包装成一个 TaskSet → 示例: RDD A → map → RDD B → map → RDD C → groupByKey → RDD D → map → RDD E → collect() 窄依赖流水线 宽依赖切Stage 窄依赖流水线 |<───── Stage 0 ──────>|<── Stage 1 ──>|<── Stage 2 ──>| Step 7: TaskScheduler 分发 Task → 接收 DAGScheduler 发来的 TaskSet → 按"数据本地性"原则分发（优先发到数据所在节点） → 监控 Task 执行状态，失败自动重试（默认 4 次） Step 8: Executor 执行 Task → 线程池里的线程执行具体的 Task → 处理完后把结果返回给 Driver → 中间数据需要缓存的写入 BlockManager Step 9: Driver 汇总结果返回用户 → 收集所有 Executor 返回的 Task 结果 → 聚合后返回给用户程序

Task 怎么生成：每个 Stage 根据 RDD 的 partition 数量拆分成多个 Task。比如 Stage 0 有 100 个 partition，就会生成 100 个 Task。

4.4 Job / Stage / Task 的关系（面试必背）

记忆口诀：

一个 Action 触发一个 Job
一个 Job 按宽依赖切成多个 Stage
一个 Stage 按分区数拆成多个 Task

一个 Application 里有多少个 Job？取决于代码里调了多少个 Action。比如：

rdd.count()// Job 1rdd.take(10)// Job 2rdd.saveAsTextFile("hdfs://...")// Job 3

这三个 Action 会依次触发 3 个 Job。

五、容错机制 —— 节点挂了怎么办

5.1 Lineage 血统重算

RDD 的每个 partition 都记录了它的"家谱"——从数据源开始，经过哪些 Transformation 一步步变过来的。如果某个 partition 的数据丢失（节点挂了），Spark 会根据这个血统信息，找到上游还活着的数据，重新计算这个 partition。

原始数据(HDFS) -> map -> filter -> reduceByKey -> 结果 ^ ^ 节点挂了！ lineage 回溯到 HDFS 重新读取 -> map -> filter -> 恢复

Lineage 的优势：不需要额外存储副本，靠逻辑记录就能恢复，节省存储。
Lineage 的劣势：如果血统链太长，重算代价太大。比如经历了 50 个 Transformation 才到这一步，重算可能要跑很久。

5.2 Checkpoint —— 切断长血统

Checkpoint 的本质是把 RDD 的当前状态持久化到可靠的分布式存储（比如 HDFS、S3），并且切断它和上游的依赖关系。

sparkContext.setCheckpointDir("hdfs:///checkpoints")vallongLineageRDD=...// 经过了很多 transformationlongLineageRDD.checkpoint()// 触发写入 HDFS

执行 Checkpoint 后，这个 RDD 的血统被截断。如果后续某个 partition 丢失，只需要从 Checkpoint 位置恢复，不需要从头重算。

5.3 Lineage 和 Checkpoint 的配合策略

实际生产中我的建议：

六、Tungsten 优化 —— Spark 3.x 的性能杀手锏

6.1 为什么要做 Tungsten？

传统 Spark（1.x 时代）的一个大问题是JVM 对象开销太大。一个int本来 4 字节就够了，但包装成Integer对象在 JVM 里要占 16 字节甚至更多。在百亿级数据处理时，内存压力巨大，GC 频繁触发，任务经常因为内存溢出被杀掉。

Tungsten 就是 Spark 团队针对这个问题做的底层重构。

6.2 UnsafeRow：把对象拍扁成二进制

Tungsten 的核心思路：用堆外内存（Off-Heap）存二进制数据，消除 JVM 对象的开销。

传统方式（JVM 对象）： Row { Integer age(16字节), String name(引用+对象), Double score(24字节) } -> 实际内存占用：100+ 字节 Tungsten UnsafeRow： [ age(4字节) | name长度(4字节) | name指针(8字节) | score(8字节) ] -> 紧凑的二进制布局，直接存堆外内存

UnsafeRow 是 Spark SQL/Dataset 的底层存储格式，它的名字里带 “Unsafe” 是因为它绕过了 JVM 的类型安全检查，直接用sun.misc.Unsafe操作内存地址，换来了极致的性能。

6.3 三个阶段（Phase 1/2/3）分别做了什么

Phase 3 是最狠的。传统执行模式下，每个算子（filter、project、join）都是一个独立的函数调用，数据在算子之间传递时要经过大量的虚函数调用和序列化。而 Whole-Stage Codegen 会把这些算子合并编译成一段连续的 Java 代码，就像手写了一个 for 循环，把 filter、map 全塞进去，编译成字节码直接执行。

你可以这样验证是否触发了 codegen：

df.explain(true)// 看执行计划里的 * 标记，带 * 的表示 codegen 后的阶段

6.4 对 RDD 用户的启示

Tungsten 主要是 DataFrame/Dataset 层面的优化，RDD 用户享受不到 Catalyst 和 Whole-Stage Codegen。但有两个点可以参考：

1. 尽量用 DataFrame 替代 RDD做结构化数据处理，性能差距可能有 10 倍以上
2. 如果必须用 RDD，注意控制 partition 内的数据量，避免单个 partition 太大导致内存压力

七、Pipeline / Operator Chain —— 什么时候断开？

7.1 流水线执行的条件

窄依赖的算子会被 Spark 合并到同一个 Task 内流水线执行。比如：

rdd.map(...).filter(...).map(...)

这三个算子会合并成一个 pipeline，在一个 Task 内完成，中间不需要把数据落盘。

7.2 什么情况下 pipeline 会断开？

7.3 实战技巧：控制 pipeline 长度

Pipeline 太长会导致单个 Task 执行时间过长，如果中途失败重算代价大；Pipeline 太短又会导致 Task 数量过多，调度开销大。

一个经验法则：

• 单个 Stage 内的 pipeline 操作控制在 5-15 个为宜
• 如果超过 15 个，考虑在中间加一个persist()，把计算成果存下来
• 如果少于 3 个，检查是否分区策略不合理，产生了过多的小 Task

八、线上常见问题与解决方案

8.1 OOM（内存溢出）—— 最头疼的问题

场景 1：单个 partition 数据量太大

报错：java.lang.OutOfMemoryError: Java heap space 或 Executor lost / Container killed by YARN

原因：某个 key 的数据高度倾斜，导致一个 partition 的数据远超其他。

解决思路：

// 1. 先给 RDD 加随机前缀打散倾斜的 keyvalprefixRdd=skewedRdd.map{case(key,value)=>valprefix=Random.nextInt(10)(s"${prefix}_$key",value)}// 2. 做聚合valaggregated=prefixRdd.reduceByKey(...)// 3. 去掉前缀二次聚合valresult=aggregated.map{case(prefixedKey,value)=>valrealKey=prefixedKey.substring(prefixedKey.indexOf("_")+1)(realKey,value)}.reduceByKey(...)

场景 2：Broadcast 变量太大

报错：Size exceeds Integer.MAX_VALUE

解决：

// 检查 broadcast 的数据大小，如果超过几 GB，改用 map-side join// 或者调整 spark.broadcast.blockSize（默认 4MB，可适当调大）spark.conf.set("spark.broadcast.blockSize","16m")

8.2 Shuffle 阶段慢 —— 数据倾斜

判断方法：看 Spark UI 的 Stage 页面，如果某个 Task 的执行时间远超其他 Task（比如其他 10 秒，这个 30 分钟），大概率是数据倾斜。

解决：

1. 加盐（Salting）：给倾斜的 key 加随机前缀，先局部聚合再全局聚合
2. 自定义 Partitioner：让数据更均匀地分布
3. 两阶段聚合：reduceByKey比groupByKey更不容易倾斜，因为reduceByKey会在 map 端先做预聚合

8.3 小文件问题

Shuffle 后如果 partition 数设置得太大，会产生大量小文件，HDFS namenode 压力大，后续读取也慢。

解决：

// 输出前合并小文件result.coalesce(10).saveAsTextFile("hdfs:///output")// 或者result.repartition(10).saveAsTextFile("hdfs:///output")

注意：coalesce只能减少分区数，不能增加；repartition可以增加也可以减少，但一定会触发 Shuffle。

8.4 Spark UI 看不到或端口冲突

本地调试时 Spark UI 默认开在 4040 端口，如果同时跑了多个 Spark 程序，端口会递增（4041、4042…）。线上环境可以通过 spark.history.server 查看历史任务的 UI。

九、生产环境调参经验

这里列几个我调过的、真正有效果的参数：

十、面试核心考点速查表

把前面讲的内容整理成面试问答形式，方便背诵：

Q：RDD 的五大特性？
A：不可变、分区、依赖关系、Transformation 懒执行、持久化。

Q：宽窄依赖区别？
A：窄依赖是父 partition 对应子 partition 一对一/多对一，不经过 Shuffle；宽依赖是父 partition 对应子 partition 一对多，必须经过 Shuffle。DAGScheduler 在宽依赖处切分 Stage。

Q：Transformation 和 Action 的区别？
A：Transformation 记录 lineage 不触发执行，返回新 RDD；Action 触发runJob，把任务提交到集群执行，返回非 RDD 结果。

Q：一个 Application 有几个 Job？
A：等于代码中 Action 的调用次数。每个 Action 触发一个 Job。

Q：Task / Stage / Job 的关系？
A：一个 Action = 一个 Job；Job 按宽依赖切分成多个 Stage；Stage 按 RDD 的分区数拆分成多个 Task。

Q：Spark 容错怎么实现？
A：主要靠 Lineage 血统重算，配合 Checkpoint 切断长 lineage。Task 失败会自动重试 4 次。

Q：Pipeline 什么时候断开？
A：宽依赖处一定断开；repartition一定断；coalesce在不触发 Shuffle 时不断。

Q：Tungsten 优化是什么？
A：Spark 的底层执行优化，包括堆外内存管理（Phase 1）、UnsafeRow 紧凑二进制格式（Phase 2）、Whole-Stage Codegen 生成字节码（Phase 3）。

写在最后

Spark Core 的内容远不止这些，但上面这些覆盖了生产环境最常用的核心知识点。建议配合 Spark UI 反复观察任务执行过程，看多了自然就理解 Stage 怎么切、Task 怎么分的了。

如果这篇文章对你有帮助，记得点赞收藏呀～