RDD源码拆解：血缘、缓存、checkpoint与调度链路

RDD 源码这东西，我前后翻过三遍。第一遍是为了搞明白一个作业为什么卡在 99% 不动，第二遍是给团队做内部技术分享，第三遍纯粹是因为线上又出了一个“同一个 RDD 被重复计算四次”的怪问题。三次翻下来最大的感受是：很多人对 RDD 的理解停留在“弹性分布式数据集”这八个字上，能背出五个特性，但一旦遇到分区倾斜、重复计算、缓存不生效、checkpoint 目录一堆空文件这类问题，就完全接不上了。这篇就按我自己的阅读顺序，把 RDD 源码从类定义一路拆到调度链路，把每个关键方法为什么这么写、这么写在生产里意味着什么讲透，顺带把那些文档里不会写的坑整理出来。适合已经能写 Spark 作业、但一遇到性能问题就只会调参数的读者，也适合想从“会用”跨到“能改”的读者。

1. 先搞清楚RDD到底替我们扛了什么事

1.1 从一次作业卡在99%的排查说起

那次出问题的作业并不复杂：读一张宽表，做几轮 join 和聚合，最后写回存储。前面 800 多个 task 三分钟跑完，最后 20 个 task 磨了四十分钟。第一反应是数据倾斜，加了随机前缀重跑，仍然卡。第二反应是资源不够，把 executor 核数从 4 提到 8，还是卡。最后把rdd.toDebugString打印出来才看明白：这条血缘链上有三个分支共用了同一个上游 RDD，而这个上游 RDD 既没有 persist，也没有被 checkpoint，于是在每个下游 task 里被重新算了一遍。表面上看是“最后几个 task 慢”，实际是这几个 task 走了最长的血缘链，把上游的读取和转换全重放了一遍。

这个故事说明一件事：Spark 的执行模型里，RDD 只描述“怎么算”，不描述“算出来的数据放在哪”。它是惰性的、不可变的、带血缘的计算说明书。真正决定性能的，是这份说明书被展开了几次、每次展开的成本有多高。想搞清楚“展开了几次”，就必须去看iterator方法和Dependency的实现；想搞清楚“展开的成本”，就必须去看compute和getPartitions。这些全都在RDD.scala这一个文件里。

所以我一直建议团队里的新人在学完map、filter、reduceByKey之后，立刻花两个小时把RDD抽象类从头读到尾。它不长，去掉注释和日志大概四百多行，但它解释了你在 Spark 上遇到的一大半性能问题。

1.2 RDD类的骨架与五个核心成员

先看类声明。在我看的这个版本里，签名大致是这样：

abstract class RDD[T: ClassTag]( @transient private var _sc: SparkContext, @transient private var deps: Seq[Dependency[_]] ) extends Serializable with Logging {

两个构造参数值得单独说。_sc标了@transient，意味着序列化 RDD 时 SparkContext 不会被一起带走——这一点很关键，因为 executor 端拿到的 RDD 对象是反序列化出来的副本，它只需要知道“怎么算”，不需要、也不能持有 driver 端的上下文。deps同样标了@transient，因为依赖关系在 driver 端构建完 Stage 后就没必要再传到执行端了，传过去只是白白增加序列化体积。

类里面真正撑起整个模型的，是五个成员：

这五条里，前三条是硬性的，后两条是优化项。我见过有人把partitioner当成纯粹的内部细节忽略掉，结果写出了“两个已经按相同 key 分区的 RDD 做 join 却还是触发 shuffle”的代码。实际上partitioner是 Spark 里少数几个能让你“白嫖”性能的地方：reduceByKey、groupByKey、join这些算子都会检查上游 RDD 的partitioner，如果发现两个输入已经用同一个Partitioner实例分过区，并且分区数一致，就直接跳过 shuffle，走mapPartitions的本地笛卡尔路径。

再补一个细节：getPartitions返回的是Array[Partition]，长度就是分区数。sc.textFile(path, 4)里的 4 最终会变成HadoopRDD里minPartitions的取值，然后由FileInputFormat按块大小切分。这里有个常见的误解：很多人以为textFile的第二个参数就是最终分区数。不是。它只是“最小分区数”的下界，实际分区数由文件的块数决定。一个 1GB 的文件按 128MB 切，会是 8 个分区，你传 4 也没用；你传 32 也没用，除非开启mapred.min.split.size之类的合并策略。这个行为在HadoopRDD.getPartitions里能清清楚楚看到。

2. compute与iterator：一个分区上的数据是怎么流出来的

2.1 MapPartitionsRDD的compute实现

理解compute最快的方式是看MapPartitionsRDD，因为map、flatMap、filter、mapPartitions这些最常用的算子最后全都落到它身上。它的核心就三行：

private[spark] class MapPartitionsRDD[U: ClassTag, T: ClassTag]( var prev: RDD[T], f: (TaskContext, Int, Iterator[T]) => Iterator[U], preservesPartitioning: Boolean = false, isFromBarrier: Boolean = false, isOrderSensitive: Boolean = false) extends RDD[U](prev) { override val partitioner: Option[Partitioner] = if (preservesPartitioning) firstParent[T].partitioner else None override def getPartitions: Array[Partition] = firstParent[T].partitions override def compute(split: Partition, context: TaskContext): Iterator[U] = f(context, split.index, firstParent[T].iterator(split, context)) }

三行分别回答了三个问题。partitioner说的是“我这个新 RDD 还算不算已分区”——只有preservesPartitioning为 true 时才继承上游，map默认是 false，因为它可能改 key；mapPartitions(..., preservesPartitioning = true)才是 true。getPartitions直接复用上游的分区数组，这说明map类算子不改变并行度，一个上游分区对应一个下游分区，一一映射。compute里最关键的是最后那个firstParent[T].iterator(split, context)——注意，它调的是上游的iterator，不是compute。

这个差别是整个惰性求值链条的核心。iterator里有一层额外的判断，会先检查缓存、再检查 checkpoint，最后才落到compute。如果上游 persist 过，这里就走 BlockManager 拿数据；如果上游 checkpoint 过，这里就转去读 checkpoint 文件；两者都没有，才真正执行上游的compute。也就是说，缓存的命中判断发生在每一层 iterator 调用上，而不是在作业开始前统一做一次。这也是为什么缓存能“穿透”多层血缘——因为每一层都在问“我有没有现成的”。

2.2 iterator里的缓存分支与checkpoint分支

把iterator的实现摊开看：

final def iterator(split: Partition, context: TaskContext): Iterator[T] = { if (storageLevel != StorageLevel.NONE) { getOrCompute(split, context) } else { computeOrReadCheckpoint(split, context) } } private[spark] def computeOrReadCheckpoint( split: Partition, context: TaskContext): Iterator[T] = { if (isCheckpointedAndMaterialized) { firstParent[T].iterator(split, context) } else { compute(split, context) } }

逻辑非常直白：有存储级别就走getOrCompute，没有就走computeOrReadCheckpoint。而computeOrReadCheckpoint里那个“读 checkpoint 竟然是调用firstParent.iterator”的写法，第一次看会觉得很怪。原因是 checkpoint 之后，RDD 的血缘被重写成了一条指向“checkpoint 文件读取 RDD”的短链——ReliableRDDCheckpointData在 checkpoint 完成时会调用rdd.markCheckpointed()，把父依赖替换成CheckpointRDD。所以firstParent此时已经指向那个读文件的 RDD 了，firstParent.iterator的语义就是“去读 checkpoint 文件”。这个设计挺巧妙的，用一次依赖重写，把“从文件读”和“从上游算”统一成了同一个调用形式，上层完全不用改。

getOrCompute则是缓存的核心：

private[spark] def getOrCompute(partition: Partition, context: TaskContext): Iterator[T] = { val blockId = RDDBlockId(id, partition.index) var readCachedBlock = true val computedValues = sc.getOrElseUpdate(blockId, storageLevel, context) { readCachedBlock = false computeOrReadCheckpoint(partition, context) } ... }

sc.getOrElseUpdate会先问 BlockManager 这个 block 存不存在：存在就直接返回反序列化后的迭代器，并且给 block 记一次“被使用”；不存在就把后面的闭包执行一遍，把算出来的迭代器写进 BlockManager，再返回。这里的重点是存储发生在 task 内部、写块的时候会受内存压力影响。如果内存不够，MEMORY_ONLY级别下这个分区会直接丢弃不存（日志里会出现Not enough space to cache rdd_X in memory），MEMORY_AND_DISK则会溢写到磁盘。这就是为什么很多人 persist 之后发现“缓存没生效”——不是没生效，是被丢掉了。

还有一点容易被忽略：getOrCompute返回的迭代器在消费完成后会触发context.addTaskCompletionListener，通知 BlockManager 这个 block 已被使用，用于 LRU 淘汰打分。也就是说，缓存的读取和释放是跟着 task 生命周期走的，如果你的 task 特别长，中间某个 RDD 缓存其实早就该被换出，它也会一直占着位置。这一点在做大 batch 迭代计算时影响很明显。

3. 依赖与血缘：窄依赖、宽依赖划出的Stage边界

3.1 Dependency家族的源码结构

Dependency本身极其简单，只有一个rdd成员：

abstract class Dependency[T] extends Serializable { def rdd: RDD[T] }

真正的分化在下面。NarrowDependency带一个getParents(partitionId)方法，回答“下游的第 N 个分区依赖上游的哪几个分区”。OneToOneDependency返回List(partitionId)，RangeDependency返回一段区间，PruneDependency处理裁剪。这三种都属于“一个下游分区只依赖有限个上游分区”，所以父 RDD 的某个分区挂了，只需要重算对应的一小段。

ShuffleDependency就完全不同了。它的签名里带着partitioner、serializer、aggregator、mapSideCombine、shuffleId：

class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag]( @transient private val _rdd: RDD[_ <: Product2[K, V]], val partitioner: Partitioner, val serializer: Serializer = SparkEnv.get.serializer, val keyOrdering: Option[Ordering[K]] = None, val aggregator: Option[Aggregator[K, V, C]] = None, val mapSideCombine: Boolean = false, val shuffleId: Int = _rdd.context.newShuffleId()) extends Dependency[Product2[K, V]]

这个类里藏着几个直接影响性能的开关。mapSideCombine决定要不要在 map 端预聚合，reduceByKey的默认实现会把它设成 true，而groupByKey是 false——这就是reduceByKey比groupByKey快得多的根源，不是因为 API 写法不同，而是因为mapSideCombine在源码层面就不一样。serializer默认取SparkEnv.get.serializer，也就是spark.serializer配置项；如果你的 key/value 是自定义对象，用默认的 Java 序列化会慢到离谱，换成 Kryo 往往能省 30% 以上的 shuffle 数据量。aggregator里带着createCombiner、mergeValue、mergeCombiners三个函数，这三个函数的实现质量直接决定聚合阶段的 CPU 开销。

还有一个细节：shuffleId是在构造时通过newShuffleId()立刻分配的，而不是等到执行时。这意味着同一个 RDD 的血缘被反复构建时，shuffleId 会不断增长。如果你在循环里无脑地创建 RDD 而不清理，shuffleId会一路涨到几万，MapOutputTracker 里会积累一堆无用的元数据，driver 内存缓慢上涨。这个问题不会报错，只会让你在跑了几十轮迭代之后发现 driver GC 时间越来越长。

3.2 getShuffleDependencies的广度优先遍历

Stage 划分的起点在DAGScheduler.getShuffleDependencies，它从结果 RDD 出发，向上找“最近的” shuffle 依赖：

private[scheduler] def getShuffleDependencies( rdd: RDD[_]): HashSet[ShuffleDependency[_, _, _]] = { val parents = new HashSet[ShuffleDependency[_, _, _]] val visited = new HashSet[RDD[_]] val waitingForVisit = new ListBuffer[RDD[_]] waitingForVisit += rdd while (waitingForVisit.nonEmpty) { val toVisit = waitingForVisit.remove(0) if (!visited(toVisit)) { visited += toVisit toVisit.dependencies.foreach { case shuffleDep: ShuffleDependency[_, _, _] => parents += shuffleDep case dependency => waitingForVisit.prepend(dependency.rdd) } } } parents }

这段代码的语义是：一路往上游走，遇到窄依赖就继续走，遇到宽依赖就把这个宽依赖记下来，但不再继续往这个宽依赖的上游走。所以它找到的是“当前 Stage 边界上的所有 shuffle”，而不是全图的 shuffle。这个设计决定了 Stage 的切分方式：一个 Stage 以读 shuffle 数据开头（或读源数据开头），以写 shuffle 数据或产出结果结尾。

visited这个集合防止了菱形结构导致的重复遍历。但要注意一个副作用：这段遍历是每个 job 提交时都要跑一遍的。如果你的血缘链特别长（比如几百层转换），而且中间有大量分支，这个遍历的耗时会累积。我见过一个自动生成的特征工程流水线，单次 job 提交阶段就要花 2 秒多，其中大半时间耗在getShuffleDependencies和getMissingParentStages上。解决办法是把血缘链截断——定期 checkpoint，或者干脆把中间结果落一次盘。

再往下就是createResultStage→getOrCreateParentStages→newStage，每个 Stage 里记录parents、rdd、numTasks、partitions。numTasks的计算来自partitions.length，也就是最后一个 RDD 的分区数。这意味着如果你的 Stage 最后一步是个coalesce(1)，那整个 Stage 就只有 1 个 task，并行度瞬间归零。这个坑在写“先聚合再写出”的流程时特别容易踩到。

4. 从行动算子到Task：调度链路上的关键一跳

4.1 runJob到handleJobSubmitted

count()这类行动算子最终会调到SparkContext.runJob：

def runJob[T, U: ClassTag]( rdd: RDD[T], func: (TaskContext, Iterator[T]) => U, partitions: Seq[Int], resultHandler: (Int, U) => Unit): Unit = { ... dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get) progressBar.foreach(_.finishAll()) rdd.doCheckpoint() }

注意最后的rdd.doCheckpoint()。checkpoint 的标记是在行动算子触发时才开始真正执行的，而且是在整个 job 跑完之后才去写文件。这解释了一个经典现象：你写了rdd.checkpoint()，然后跑了一个count()，发现 checkpoint 目录里啥也没有。因为count的 job 结束后doCheckpoint确实会启动一个写文件的 job，但那个 job 是异步的、有延迟的，而且它是从头算一次再写——不是复用刚才count算出来的结果。想让 checkpoint 复用缓存，必须按“先 persist、再 checkpoint、最后触发”的顺序写。

dagScheduler.runJob内部会创建JobWaiter，然后submitJob，把JobSubmitted事件投递到DAGSchedulerEventProcessLoop。事件循环是单线程的，所有 Stage 划分、task 提交、失败处理都在这个线程里串行完成。这就带来一个隐性约束：driver 端的调度逻辑不能有阻塞调用。如果你的RDD子类在getPartitions里做了耗时的 IO（比如查元数据库），整个调度线程都会被卡住，所有并发 job 一起排队。这个坑在自定义数据源时非常常见。

4.2 submitMissingTasks与任务本地性

Stage 建好之后进入submitMissingTasks。这里会做几件事：算出partitionsToCompute（还没算完的分区）、序列化 task 的二进制、根据preferredLocations算出每个 task 的本地性级别，然后打包成TaskSet交给TaskSchedulerImpl。

本地性级别有这么几档，从好到坏是PROCESS_LOCAL、NODE_LOCAL、NO_PREF、RACK_LOCAL、ANY。Spark 会先按最好的级别试着调度，如果spark.locality.wait时间内没等到资源，就降一级再试。默认是 3 秒。这个参数调大能让本地性更好，但也可能让集群空转；调小能提高资源利用率，但会增加跨节点读数据的开销。

这里面有个反直觉的地方：getPreferredLocations返回的 host 只是“提示”，不是“约束”。如果那个节点没有空闲 slot，task 就会被发到别处，然后从那个节点远程拉数据。对于HadoopRDD来说，getPreferredLocations返回的是 HDFS 块所在的 datanode 列表；对于 shuffle 之后的 RDD，返回的是MapOutputTracker里记录的 map 输出位置。所以 shuffle 之后如果分区数配得太大，每个分区的数据量很小，网络传输的开销占比就会急剧上升——不是数据量的问题，是“每个 task 都要建立一次连接去拉几个 KB”的问题。

关于分区数的计算，我一般这么估：总输入数据量除以目标分区大小。假设一次读取 200GB 的 Parquet，目标单分区 128MB 到 256MB，那么200 * 1024 / 128 ≈ 1600个分区。如果 executor 有 20 个，每个 4 核，总共 80 个 slot，1600 个 task 大概跑 20 轮，每轮 1 到 2 分钟，整体可控。反过来如果只开 200 个分区，单分区 1GB，很容易出现 executor 内存吃紧、GC 时间飙升；如果开 8000 个分区，单分区 25MB，调度开销和 task 启动开销占比就会很高。这个计算没什么玄学，就是除法，但很多人调参时不去算，直接拍脑袋写个数，然后就一直在“倾斜”和“OOM”之间来回横跳。

5. 缓存与checkpoint：容错链条上的两道保险

5.1 persist与StorageLevel的实现细节

persist的实现不长，但有几处陷阱：

private def persist(newLevel: StorageLevel, allowOverride: Boolean): this.type = { if (storageLevel != StorageLevel.NONE && newLevel != storageLevel && !allowOverride) { throw new UnsupportedOperationException( "Cannot change storage level of an RDD after it was already assigned a level") } ... storageLevel = newLevel this }

第一，存储级别一旦设定就不能改，除非先unpersist。所以你如果在公共方法里给某个 RDD persist 了MEMORY_ONLY，调用方再想改成MEMORY_AND_DISK_SER，直接抛异常。这种设计是为了防止已经写进去的 block 格式和新的级别对不上，但从使用者角度看确实有点硬。

第二，persist本身不触发任何计算。它只是把storageLevel这个字段改了，真正写数据要等到iterator被调用、getOrCompute执行。所以“persist 之后立刻看存储页面发现没有缓存”是完全正常的，你得有一个行动算子去触发。

第三，StorageLevel那几个布尔值的组合有讲究。MEMORY_ONLY是useMemory=true, useDisk=false, deserialized=true, replication=1；带_SER后缀的是deserialized=false，也就是存序列化后的字节数组。序列化存储的好处是省内存、省 GC 压力，坏处是每次读都要反序列化。我的经验是：对象字段多、每次只用到其中一小部分时用_SER；对象本身很小、访问频率极高时用反序列化形式。另外_2后缀是双副本，会占双倍内存，但在节点故障频繁的环境里能省掉重算时间，值不值要看你那边节点稳定性。

顺手提一个内存估算的例子。假设一个 executor 配了 8GB 堆，spark.memory.fraction默认 0.6，就是 4.8GB 给统一内存池，其中 storage 和 execution 各占一半（spark.memory.storageFraction默认 0.5），也就是 2.4GB 理论上可用于缓存。但 execution 可以抢占 storage，所以实际可用往往不到 2.4GB。如果你的一个分区反序列化后有 200MB，那最多也就缓存十来个分区，跟总分区数一比就知道“能不能全缓存住”。算一下就知道该不该指望缓存，比盲目 persist 再抱怨“缓存命中率低”靠谱得多。

5.2 checkpoint如何截断血缘

checkpoint()方法本身只做两件事，检查目录配置、挂上ReliableRDDCheckpointData：

final def checkpoint(): Unit = RDDCheckpointData.synchronized { if (context.checkpointDir.isEmpty) { throw new SparkException("Checkpoint directory has not been set in the SparkContext") } else if (checkpointData.isEmpty) { checkpointData = Some(new ReliableRDDCheckpointData(this)) } }

真正干活在doCheckpoint，它被行动算子调用，内部会起一个 job 把数据写到checkpointDir下，写完之后调用markCheckpointed()，把依赖重写掉。重写之后，这个 RDD 的血缘长度从“整条上游链”缩短成“一个 CheckpointRDD”，重新计算时只需要读文件。

这里有个非常重要的实践点：checkpoint 之前必须 persist，否则写文件那次会重算一遍。流程是“action 触发 → 算一遍 → 写 checkpoint 文件”，如果中间没有缓存，写文件时又要把上游全算一遍。这就是为什么官方文档反复强调这个顺序，也是为什么很多人 checkpoint 之后发现作业时间翻倍。

还有一个容易混淆的概念是localCheckpoint()。它不写可靠存储，只是把数据缓存起来并把血缘截断到当前 RDD，等价于“persist + 切断依赖”。好处是快，不需要额外 IO；坏处是 executor 挂了数据就没了，而且它还占着缓存空间。在长链路迭代计算里，localCheckpoint 往往比完整 checkpoint 更实用。但要注意，用了 localCheckpoint 之后这个 RDD 就不能再被别的 job 当普通 RDD 用了，它的依赖已经被改写成空，读到的是缓存块，缓存一丢就彻底没了。这个坑我在一次多 job 共享中间结果的场景里踩得很实，第二个 job 直接抛了“block not found”。

6. 实战中的坑与排查手法

6.1 常见问题速查表

把上面这些源码层面的理解翻译成排查动作，大概是这样一张表：

这张表我基本是贴在工位上的，遇到问题先对一遍，能省掉一半的瞎猜时间。

6.2 几个我踩过的具体坑

第一个坑是coalesce和repartition的混淆。coalesce(1)不触发 shuffle，它只是把多个上游分区合并到一个 task 里顺序读，所以上游有多少分区，这个 task 就要读多少份数据，单个 task 的内存压力是上游总和。而repartition(1)走的是 shuffle，会把数据真正打散重分。这两个在源码里的路径完全不同：前者是CoalescedRDD，getPartitions直接算出合并后的分区映射；后者是ShuffledRDD，带一个ShuffleDependency。写“最后输出一个文件”的代码时，如果数据量大，coalesce(1)极容易 OOM，而repartition(1)虽然多一轮 shuffle，但内存曲线平稳得多。

第二个坑是自定义Partitioner时忘了实现equals和hashCode。ShuffledRDD判断能否复用上游分区时，比较的是partitioner对象。如果你每次都new一个新实例又不重写equals，那么即使逻辑完全一样，也会被判为不同，白白触发 shuffle。这个问题的隐蔽之处在于它不报错，只是慢。我当时的做法是改成单例对象，或者干脆用HashPartitioner这类内置实现。

第三个坑跟mapPartitions的迭代器消费有关。compute返回的是Iterator，很多人在里面写“先遍历一遍统计数量，再遍历一遍处理数据”，结果发现第二遍是空的。因为Iterator是一次性的，消费完就到底了。正确做法是toList或者用mapPartitionsWithIndex配合一次遍历完成。这个错误在写自定义聚合逻辑时特别常见，而且因为不报错、只是结果为空，排查起来很费劲。

第四个坑是cache的位置。很多人习惯在val rdd = sc.textFile(...)之后立刻.cache()，但如果后面紧跟着filter把数据量砍掉了 90%，那缓存的就是没过滤的大表，白占内存。正确的位置是在“会被多次使用的那个 RDD 上”，而不是“拿到的第一个 RDD 上”。判断标准很简单：看toDebugString里哪个节点有分叉，就在分叉点之前缓存。这个判断动作花不了十秒，但能省下大量内存。

最后再分享一个我自己常用的手法：调试阶段在关键节点插一行println(rdd.toDebugString)。它会打印出带缩进的血缘树，每个 RDD 后面跟着分区数、存储级别和分区器信息。靠这一行输出，我基本能判断出“这个作业会不会重复计算”“缓存到底挂在哪个节点上”“shuffle 发生了几次”。这比翻 Spark UI 快得多，尤其是在作业还没跑起来、只想先看看执行计划的时候。等你把血缘树看熟了，再回头看RDD.scala里那些@transient和lazy val，会发现它们每一个都在解决一个具体的问题，而不是随便加的修饰符。