Spark RDD源码解析：五大属性、Stage切分与性能调优实战

1. 先把 RDD 当抽象类看一次源码视角的整体拆解1.1 为什么还要回头啃 RDD 这段几百行的源码RDD 源码解析这件事我身边不少做数据开发的朋友第一反应是“这玩意儿不是早就会用了吗”。会用map、filter、reduceByKey确实能撑起日常八成需求。但只要遇到分区数不对、stage 卡在某个百分比起不来、checkpoint 之后任务反而跑了两遍、缓存占了内存却好像没生效这些情况光靠 API 文档里的那几句话根本解释不清。RDD 源码解析实战的价值就在这里它把一个“看起来只是集合”的东西还原成一套有明确契约的抽象类你顺着这份契约往下看很多现场问题就不再是玄学。RDD 是 Spark 最核心的数据抽象这个抽象类本身只有几百行但它定义了五条硬约束分区列表、分区计算函数、依赖关系、分区器、优先位置。整个调度层、存储层、容错层都是围绕这五条约束展开的。你把这几百行读透等于拿到了一张地契后面看 DAGScheduler 怎么切 stage、BlockManager 怎么存块、Task 怎么被派发都是在你的地契上盖房子。这篇内容适合两类人一类是能写 Spark 作业但对内部机制只有模糊印象的开发者另一类是被线上任务折磨过、想搞清楚“为什么这么慢”的调优负责人。我会从抽象类的结构一路拆到 stage 切分的调度链路中间穿插自己踩过的坑和几段可以直接抄的自定义 RDD 代码。读源码这件事不需要你把 Scala 玩到多溜能看懂abstract class、extends、override就够了。1.2 版本锚定不同版本源码差异比你想的大开始之前先做一件事把版本钉死。Spark 2.4、3.0、3.3、3.5 这几个节点上RDD 的核心抽象基本稳定但外围的调度细节、shuffle 写文件实现、自适应执行相关的改动不少。我一般的做法是下载对应版本的源码包用 IDE 打开core/src/main/scala/org/apache/spark/rdd/RDD.scala先在注释区扫一遍再从头读方法定义。版本如果不锚定你在网上搜到的分析文章很可能讲的是 1.6 时代的东西跟手上的运行时对不上读完反而更乱。我自己的习惯是同时开三个窗口一个是 RDD.scala一个是Dependency.scala一个是DAGScheduler.scala。前两个决定“数据怎么算”后一个决定“任务怎么切”。这三份文件加起来不到一万行真正跟主干逻辑相关的可能就两千行通读一遍一个下午够了。看的时候别追求一次看懂所有分支第一次只跟主干iterator怎么调computedependencies怎么被getShuffleDependencies消费partitions怎么被partitionsToCompute用上。还有一个容易被忽略的点泛型上的ClassTag。abstract class RDD[T: ClassTag]这个上下文界定不是装饰它保证运行期能拿到类型信息序列化、数组创建、Kryo 注册都要用它。你写自定义 RDD 时如果类型参数没带ClassTag编译期就会卡住这是新手最常撞的第一堵墙。1.3 一个被低估的设计取舍不可变加血统带来的重算成本RDD 被设计成不可变、可追溯血统的结构这个选择在源码里体现得非常直白它不存数据只存“怎么算出数据”的配方。compute方法每次被调用都是从父 RDD 的迭代器重新拉一遍。这个设计的优点是容错极其廉价——某个分区丢了照着血统从最近的可用点重算就行不需要副本、不需要日志。代价也很明显一条长长的窄依赖链如果中间某个 RDD 被多次复用却没有缓存那它上游的所有计算都会被重复执行。我在一个日志清洗任务里吃过这个亏。当时链路是“读文件 → 解析 → 过滤 → 关联维表 → 聚合”关联维表那一步是一个中等开销的 broadcast join结果我在下游写了两个 action一个落库一个统计没做任何缓存。跑起来发现关联那步执行了两遍整体耗时直接翻倍。回头看书iterator里只有storageLevel ! StorageLevel.NONE时才会走getOrCompute否则老老实实computeOrReadCheckpoint。源码写得清清楚楚只是以前没看。所以读 RDD 源码第一个要建立的直觉是它不是数据容器是计算说明书。这句话听着像口号但你把它代入到每一次 action 触发的执行里很多性能问题就有了统一的解释框架。2. 五大属性逐行对照抽象方法与它们的真实约束2.1 抽象类的骨架哪些必须实现哪些可以选择性覆盖把 RDD 的类声明摊开看最关键的是这几个成员。第一个是构造函数里的SparkContext和deps前者标记这个 RDD 属于哪个上下文后者是依赖列表的初始值第二个是compute(split, context)返回一个迭代器这是唯一的必选计算入口第三个是getPartitions返回分区数组第四个是getDependencies默认返回构造时传入的依赖第五个是getPreferredLocations默认返回空序列。abstract class RDD[T: ClassTag]( transient private var _sc: SparkContext, transient private var deps: Seq[Dependency[_]] ) extends Serializable with Logging { DeveloperApi def compute(split: Partition, context: TaskContext): Iterator[T] protected def getPartitions: Array[Partition] protected def getDependencies: Seq[Dependency[_]] deps protected def getPreferredLocations(split: Partition): Seq[String] Nil transient val partitioner: Option[Partitioner] None }注意transient这个注解它标在_sc和deps上意思是序列化时跳过。因为 RDD 会被闭包捕获带到 executor 上如果把整个SparkContext序列化过去体积和语义都会出问题。executor 端拿到的是一个“反序列化后字段为 null”的 RDD 外壳靠Task里携带的上下文重建执行环境。这也是为什么你在compute里绝对不要直接引用sc做动作那在 driver 上能跑在 executor 上就是空指针。这个坑我见过不止一次有人图省事在map里调sc.broadcast本地模式一切正常一上集群就崩。partitioner是val声明为Option[Partitioner]默认None。只有像ShuffledRDD、ParallelCollectionRDD这类明确知道分区规则的具体实现才会覆盖它。这个字段是 Spark 优化reduceByKey、join时避免多余 shuffle 的关键依据两个 RDD 分区器一致且分区数相同join 就可以免 shuffle 直接按分区对齐。2.2 getPartitions 与分区数一个 require 检查能省你半天partitions这个 final 方法比getPartitions更值得看因为它在调用抽象方法之外还做了两件事缓存结果、校验下标。final def partitions: Array[Partition] { checkpointRDD.map(_.partitions).getOrElse { if (partitions_ null) { partitions_ getPartitions partitions_.zipWithIndex.foreach { case (partition, index) require(partition.index index, sAttempting to access a partition with index $index, but the RDD has s${partitions_.length} partitions numbered 0 to ${partitions_.length - 1}.) } } partitions_ } }那段require是我认为整个文件里最有性价比的几行代码。它强制要求getPartitions返回数组里第 i 个元素的index必须等于 i。你写自定义 RDD 时如果分区编号从 1 开始或者过滤掉了一些分区导致下标不连续就会直接抛异常。报错信息还算友好但第一次遇到时容易懵因为异常出现在partitions被访问的时刻可能离你写getPartitions的地方隔了好几个 stage。分区数从哪来取决于具体的 RDD 实现。读 HDFS 文件的HadoopRDD分区数由文件切分决定切分大小是max(minSize, min(goalSize, blockSize))其中goalSize totalSize / minPartitionsminPartitions就是你传进去的那个参数不传则用defaultParallelism。本地集合用parallelize时分区数就是你显式给的numSlices不给则是defaultParallelism。这里有个常被问到的数值local[*]模式下defaultParallelism等于可用核数集群模式粗粒度调度器下等于spark.default.parallelism没配的话取 executor 总数乘以每个 executor 的核数且不低于 2。我做容量评估时有个经验公式单个分区处理的数据量控制在 128MB 到 256MB 比较舒服低于 64MB 说明分区可能过碎调度开销占比会上升高于 512MB 则单 task 的 GC 压力和失败重算成本都会变高。这个不是硬标准但对绝大多数按行处理的 ETL 场景都适用。2.3 compute 与迭代器模型为什么 Spark 不会中间落盘compute返回的是Iterator[T]而不是集合这个签名决定了 Spark 的执行模型。整个窄依赖链上每个 RDD 的compute拿到的是父 RDD 的迭代器套一层自己的转换逻辑再返回。数据像流水一样从源头一路推到 action中间不物化。这就是所谓的 pipeline 或者迭代器融合。我常用一个类比解释给新人做菜的时候洗菜、切菜、下锅是一个连续动作不需要每洗完一棵菜就装盘放冰箱再拿出来切。只有遇到必须换工序的地方——比如要把所有食材先集中起来按类别归堆——才需要把中间结果落到台面上。Spark 里的这个“落台面”就是 shuffle。iterator方法本身很短但它把存储层的判断插了进来final def iterator(split: Partition, context: TaskContext): Iterator[T] { if (storageLevel ! StorageLevel.NONE) { getOrCompute(split, context) } else { computeOrReadCheckpoint(split, context) } }getOrCompute会去 BlockManager 里找这个分区对应的块找到就直接返回反序列化后的迭代器找不到才调computeOrReadCheckpoint算一遍然后把结果写进去。computeOrReadCheckpoint则优先看有没有 checkpoint有就读 checkpoint 文件没有才真正调compute。这里有个实操细节值得强调迭代器模型下异常的发生时机被推迟了。你在map里写了个会抛异常的解析逻辑这个异常不会在map调用时出现而是等到 action 拉取数据、迭代器被消费到那一条记录时才炸出来。调试的时候堆栈会指向Task.run附近看着跟你的业务代码没关系。我的应对办法是在转换逻辑里加一层带行号的包装或者用mapPartitions手动 try-catch 并记录分区号和偏移量定位效率高很多。2.4 依赖与分区器窄宽之分决定了后面一切getDependencies默认返回构造时传入的deps但真正被调度层使用的是 final 的dependencies方法它在 checkpoint 之后会返回一个指向 checkpoint RDD 的单亲依赖等于把血统截断。依赖分两类源码里的类名很直白NarrowDependency和ShuffleDependency。窄依赖的核心方法是getParents(partitionId)返回子分区依赖了父 RDD 的哪些分区。OneToOneDependency直接返回List(partitionId)一一对应RangeDependency返回一个区间用于union这类操作。判断标准就一条一个子分区是否只依赖固定数量的父分区。是就是窄的可以在同一个 task 里用流水线算完否就是宽的必须落盘再做跨节点拉取。ShuffleDependency的构造函数里有个动作值得注意它在实例化时就去shuffleManager注册了一次 shuffle拿到shuffleId和shuffleHandle。这意味着依赖关系一旦建立shuffle 的元信息就已经在 driver 上登记好了后面ShuffleMapTask写文件、ShuffledRDD读文件都靠这个 handle 找到对应目录。也正因为注册发生在构造期你在map里动态创建 RDD 属于极其危险的操作——那是 driver 端的逻辑别在闭包里做。partitioner的另一个作用是决定 shuffle 之后的分区归属。HashPartitioner用 key 的哈希取模RangePartitioner通过采样确定边界。分区器不一致时join之类的操作就必须走 shuffle源码里PartitionerAwareUnionRDD、ShuffledRDD这些类的存在就是为了处理各种分区器组合。我在排查“为什么这个 join 没走 map 端优化”时第一步永远是打印两个 RDD 的partitioner和partitions.length十次有八次问题出在这。2.5 getPreferredLocations数据本地性是怎么被传递上去的这个方法的默认实现返回Nil看着无关紧要实际上它决定了 task 被派发到哪台机器。HadoopRDD会返回文件块所在的 host 列表BlockRDD会返回缓存块所在的 executor。调度层拿到的不是 RDD 自己的返回值而是DAGScheduler.getPreferredLocs递归向上查找的结果它会沿着窄依赖一路往上找直到碰到 shuffle 边界为止把最上游的位置信息带下来。为什么只在窄依赖范围内传递因为宽依赖之后数据已经被打散重分布了上游的位置信息没有意义。这个设计直接影响性能如果你的链路是“读文件 → 一堆 map → shuffle → 聚合”那么 shuffle 之前的 stage 能享受数据本地性task 会被尽量派到数据所在节点shuffle 之后就只能靠调度器的普通分配策略了。实际操作中如果发现 stage 里大量 task 的本地性级别是ANY而不是NODE_LOCAL通常意味着资源不足或者数据分布太散。可以调spark.locality.wait系列参数让调度器多等一会儿再降级。但我不建议把这个值调得太大等待时间过长反而拖慢整体默认的 3 秒对多数场景够用了。3. 从 action 到 TaskDAGScheduler 切 stage 的源码链路3.1 runJob 的调用链一次 action 触发了什么从collect或者count这类 action 点进去很快就到SparkContext.runJob。它做的事情是把你的函数、目标分区、结果处理器打包交给DAGScheduler.runJob后者创建一个JobWaiter并投递JobSubmitted事件到调度器的消息循环里然后阻塞等待完成。这段代码本身没什么复杂的复杂的是它背后的异步模型。DAGScheduler是单线程事件循环所有 stage 切分、task 提交、完成回调都在这个线程里串行处理。这个设计的好处是状态管理简单不用到处加锁坏处是如果某个事件处理耗时过长整个调度都会被阻塞。我遇到过一次 driver 端在handleTaskCompletion里做大量反序列化导致调度延迟的情况表现是所有 task 都跑完了但 job 迟迟不结束最后定位到是结果集太大、JobWaiter聚合时压力过大。解决办法是不要在 action 里collect大结果改成写文件或者分批取。runJob还有一个容易忽略的参数是partitions。默认是0 until rdd.partitions.length也就是每个分区一个 task。有些场景你可以只对部分分区执行 action比如采样或者分段校验这时候传入子集就行源码层面完全支持只是平时用不到。3.2 stage 切分的递归逻辑从最后一个 RDD 往上爬真正有意思的是handleJobSubmitted到submitStage这段。核心思路是从最终 RDD 出发找出它最近的 shuffle 依赖每一个 shuffle 依赖的父 RDD 及其上游构成一个父 stage递归处理直到没有 shuffle 依赖为止然后从最上游的 stage 开始提交。找最近 shuffle 依赖用的是getShuffleDependencies一个广度优先遍历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 }关键在case shuffleDep那个分支碰到 shuffle 依赖只记录下来不把它的父 RDD 放进待访问队列。这意味着遍历会在 shuffle 处“停住”返回的是这一层最近的 shuffle 边界集合。父 stage 的构建再对每个 shuffle 的父 RDD 递归调用同样的逻辑一层层往上直到某个 RDD 没有任何 shuffle 依赖那就是最源头的 stage。这个递归结构解释了为什么 stage 的数量等于整条链路里 shuffle 的次数加一。也解释了为什么减少不必要的 shuffle 是性能优化的第一原则——每一次 shuffle 不只是数据搬运的代价还多了一层 stage 调度、多了一轮 task 启动、多了一份落盘文件。3.3 宽依赖落盘的源码线索shuffle 写到底写了什么父 stage 的每个分区对应一个ShuffleMapTask它在 executor 上执行时会调用ShuffleDependency关联的 writer 处理器。选择哪种 writer 由三个条件决定是否需要 map 端聚合、分区数是否小于阈值、序列化器是否支持 relocation。默认配置下小分区数场景走BypassMergeSortShuffleWriter它给每个分区开一个临时文件直接写最后合并大分区数走SortShuffleWriter或者UnsafeShuffleWriter先写一份数据文件再写一份索引文件索引记录每个分区的偏移量。这个实现细节的价值在于解释现象。比如你在日志里看到大量临时文件那是 bypass 模式的中间产物比如 reduce 端拉取时报 “Failed to connect” 或者 fetch 失败往往是某个 map 输出过大导致单文件超过限制。相关参数像spark.shuffle.file.buffer控制写缓冲spark.reducer.maxSizeInFlight控制单次拉取大小spark.shuffle.io.maxRetries控制重试次数。我一般会在 shuffle 数据量超过几十 GB 的作业里把拉取大小从默认 48MB 调小到 24MB 左右牺牲一点吞吐换稳定性fetch 失败率明显下降。下游的ShuffledRDD.compute则是反方向根据shuffleHandle拿到 reader读取属于自己的那一段。reduceByKey之所以比groupByKey省内存源码层面的原因就在ShuffleDependency的mapSideCombine标志上。前者为 truemap 端会先用 combiner 做局部聚合再落盘写出去的数据量可能只有原来的几分之一后者为 false所有原始记录原样落盘reduce 端才做归并内存压力全压在下游。4. 迭代器计算模型与缓存、checkpoint 的真相4.1 persist 的存储级别内存里到底存了什么persist的入口在 RDD 上它做两件事更新storageLevel字段把当前 RDD 注册到SparkContext的持久化列表里。真正的存储在getOrCompute里发生调用BlockManager.getOrElseUpdate键是RDDBlockId(rddId, partitionIndex)。存储级别决定数据以什么形态落地。MEMORY_ONLY存反序列化对象读的时候最快但占内存最多且对 GC 不友好MEMORY_ONLY_SER存字节数组省内存但要反序列化MEMORY_AND_DISK在内存不够时溢写到磁盘DISK_ONLY直接落盘。源码里StorageLevel是一个带多个布尔标志的 case classuseMemory、useDisk、deserialized、replication这几个字段组合出所有级别。我踩过最深的一个坑是MEMORY_ONLY加一个超大对象。当时缓存的是一个包含大量嵌套结构的 case class 集合分区数量又少结果 executor 内存被打满大量时间花在 GC 上任务反而比不缓存还慢。后来换成MEMORY_ONLY_SER并配合 Kryo 序列化内存占用降到三分之一左右速度才回来。结论是缓存不是免费的级别选错等于给自己挖坑。还有一个必须注意的点persist是惰性的它只是打个标记第一次 action 执行时才真正填充数据。而且缓存的数据不会自动释放除非你显式unpersist或者上下文结束。长时间运行的应用里缓存了一堆临时 RDD 却不释放是内存泄漏的常见来源。我现在的习惯是给每个缓存点写注释说明用途和预期生命周期任务结束前统一清理。4.2 checkpoint 的两遍计算为什么会跑两次checkpoint的源码路径比较绕值得单独讲。调用rdd.checkpoint()时它只是设置了一个标记并要求你事先配置好 checkpoint 目录否则抛异常。真正的动作发生在 job 执行完成之后SparkContext.runJob在JobWaiter完成时调用rdd.doCheckpoint()后者触发一个额外的 job把当前 RDD 的数据重新算一遍并写到可靠存储里然后构造一个指向该目录的ReliableCheckpointRDD替换掉原来的血统。这就是“checkpoint 会跑两遍”的根源。第一遍是你原本的 action第二遍是 checkpoint 自己的物化 job。如果这个 RDD 上游计算很贵总耗时直接翻倍。官方建议是在 checkpoint 之前先persist这样第二遍可以直接读缓存不用从头重算。rdd.persist(StorageLevel.MEMORY_AND_DISK_SER) rdd.checkpoint() // 触发一次 action完成第一遍计算和 checkpoint 物化 rdd.count() rdd.unpersist()这段顺序不能乱。先 checkpoint 后 persist 的话第二遍计算时缓存是空的等于白干。另外 checkpoint 完成之后血统被切断如果后续发现上游有 bug 需要重跑就只能从 checkpoint 文件重新开始这在一定程度上牺牲了灵活性。我的经验是只在链路特别长、重算代价特别高、且数据本身已经稳定的场景才用 checkpoint日常用缓存就够了。4.3 失败重算的代价估算血统越长越贵task 失败时的重算逻辑在TaskSetManager里它会重新提交失败的分区executor 上重新执行一遍compute链。窄依赖链上重算的范围就是这个分区及其上游所有未被缓存的中间结果如果上游有 shuffle那么从这个 shuffle 的输出开始重算即可因为 shuffle 文件是落盘的不需要重跑 map 端除非 map 端输出也丢了。这就引出一个实用的判断shuffle 天然是一道容错屏障。链路上有多个 shuffle 时失败重算的代价被限制在最近的 shuffle 之后。反过来一条超长的窄依赖链如果中间没有任何缓存和 shuffle一个分区失败就要从源头重新读文件、重新解析、重新计算代价可能是几分钟。我做过一次粗略测算假设上游有 5 个 map 阶段每个阶段单分区处理耗时 10 秒那么失败一次重算大约 50 秒如果中间缓存了一个关键节点重算时间可以压到 20 秒以内如果在关键节点做了 checkpoint甚至可以接近 0。这个账算清楚之后你就知道该在哪里加缓存而不是凭感觉到处cache。失败重试次数由spark.task.maxFailures控制默认 4 次。有些团队为了追求稳定把它调大到 10但如果失败原因是数据本身的脏数据或者代码 bug调大只是让作业跑得更久才失败。我更倾向保持在默认值附近把精力放在定位根因上。5. 动手写一个自定义 RDD把原理变成可跑的代码5.1 需求设定从一批分片数据源读取假设你有一个自定义数据源数据按编号分片存放在若干位置每个分片需要单独读取读取过程有 IO 开销希望能并行处理并且希望能利用数据所在位置做本地性调度。这种场景下用parallelize把数据先拉到 driver 再分发是错的会撑爆内存正确的做法是实现一个自定义 RDD把分片读取逻辑下沉到 executor。先定义分区类注意index必须从 0 开始连续class ShardPartition(val index: Int, val shardId: String, val location: String) extends Partition再实现 RDD 主体核心是三个方法getPartitions返回分区数组compute返回迭代器getPreferredLocations返回位置信息。注意getPartitions和getPreferredLocations是protectedcompute是公开的。class ShardRDD( sc: SparkContext, shards: Seq[(String, String)] ) extends RDD[String](sc, Nil) { override protected def getPartitions: Array[Partition] { shards.zipWithIndex.map { case ((shardId, loc), idx) new ShardPartition(idx, shardId, loc).asInstanceOf[Partition] }.toArray } override def compute(split: Partition, context: TaskContext): Iterator[String] { val p split.asInstanceOf[ShardPartition] new ShardReader(p.shardId).readLines() } override protected def getPreferredLocations(split: Partition): Seq[String] { Seq(split.asInstanceOf[ShardPartition].location) } }继承时传Nil作为依赖表示它是源头 RDD没有父 RDD。ShardReader是你自己封装的读取器返回一个惰性迭代器一次读一行不要一次性readAll再转迭代器那样就失去了流式的意义。5.2 注册与使用类型信息和序列化要一起考虑自定义 RDD 的类必须可序列化因为它会被闭包带到 executor。大多数情况下继承 RDD 就自动带上了Serializable但你的构造参数里如果引用了不可序列化的对象比如数据库连接、文件句柄就会在提交任务时炸掉。解决办法是把这类资源声明为transient在compute里按需创建用完关闭或者用单例加懒加载的方式管理。使用方式很直接val shards Seq((shard-001, host-a), (shard-002, host-b)) val rdd new ShardRDD(sc, shards) val count rdd.map(_.trim).filter(_.nonEmpty).count()跑之前建议先做一次小规模验证只放一个分片、几行数据确认getPartitions返回的数组下标连续、compute返回的迭代器能正常消费完、异常能正确抛出。我见过有人把分片编号当成Partition.index用结果partitions那个require检查直接报错报错信息指向的却是调度阶段排查花了不少时间。如果数据源支持按范围切分还可以实现getPartitions时按数据量动态决定分片数思路跟读文件的切分算法一样总量除以目标大小向上取整最后一片兜底剩余部分。这样分区数就跟着数据量走不用手工指定。5.3 常见坑位自定义 RDD 最容易翻车的三个地方第一个坑是分区数零。getPartitions返回空数组时RDD 的分区数是 0后续 action 得到的 task 数也是 0结果是空结果而不是报错。如果你的数据源在某种条件下没有分片最好主动抛异常或者补一个空分区别让它静默返回空。第二个坑是compute返回的迭代器不幂等。同一个分区可能因为失败重试被计算多次如果compute内部有副作用比如写日志文件、更新计数器重试时就会重复执行。源码里没有任何机制帮你做去重这是调用方必须自己保证的。我的做法是副作用操作全部挪到 action 之后compute里只做纯粹的数据读取和转换。第三个坑是位置信息造假。getPreferredLocations返回的 host 必须是真实的 executor 所在主机名如果随便填一个调度器会尝试把 task 派过去找不到就降级等待白白浪费时间。本地模式下调这个方法基本无效因为它只有一个 executor任何位置都会被认为是本地的。6. 排查实录从源码视角定位那几类最烦人的问题6.1 分区数设置一个数字牵动整条链路分区数是最容易被随手设置又最容易出问题的参数。太少单 task 处理数据量大GC 时间长失败重算代价高shuffle 写文件也可能出现单文件超大太多task 启动和调度开销占比上升每个 task 处理几百 KB 数据几万个 task 的调度本身就要花好几分钟。我的经验是从数据量反推先估算总输入大小按单分区 128MB 到 256MB 目标算出基础分区数再结合集群核数做调整保证分区数至少是总核数的两到三倍这样能有足够的并行度也不至于让每个 task 太碎。如果链路里有聚合类操作shuffle 后的分区数可以适当减少因为聚合后数据量通常会显著下降。还有一个隐性的坑读文件时minPartitions参数不生效的情况。如果文件本身很小、块数很少即使你传了很大的minPartitions实际分区数也上不去因为切分算法不会把一个小文件硬拆成多片。这时候要么在读取后手动repartition要么在源头做合并。反过来大量小文件会导致分区数过多每个分区处理一个几十 KB 的文件调度开销占大头需要先用合并的手段预处理。6.2 内存溢出与序列化从报错位置反推根因OOM 的报错信息通常指向 executor 上某个具体的类但真正的根因可能在很远的代码里。我一般的排查顺序是先看是 driver 端还是 executor 端再判断是堆内还是堆外最后定位到具体的算子。driver 端 OOM 最常见的原因是collect把大结果拉回来了。源码层面runJob的结果会累积在JobWaiter里数据量超过 driver 堆就直接炸。解决办法是把结果写到分布式存储或者用take取样本、用foreachPartition在 executor 侧做聚合。executor 端 OOM 常见于三种情况一是 shuffle 读端一次性加载太多数据做聚合groupByKey是典型代表换成reduceByKey或aggregateByKey能缓解二是缓存级别选错反序列化对象占内存过多三是单条记录本身超大比如一行 JSON 几百 MB任何算子都会出问题需要在读取后立刻过滤或截断。序列化问题表现为NotSerializableException报错会指出具体是哪个字段。源码里的闭包清理机制会尽量裁剪不需要的引用但如果你在匿名函数里捕获了外部的不可序列化对象它还是会跟着走。常见的有数据库连接、日志句柄、SparkContext本身。解决办法是把这些引用声明为transient并在使用处重新创建或者用broadcast分发小对象。6.3 任务卡在某个百分比stage 边界的典型症状一个 stage 里所有 task 都跑完了但进度条停在 99% 不动多半是某个 task 在做最后的落盘或者结果汇总。如果停在中间某个位置长时间不动可能是数据倾斜少量 task 处理了绝大部分数据。从源码角度看stage 的完成条件是所有 task 都成功任何一个慢 task 都会拖住整个 stage。定位方式是看 Spark UI 里的 task 分布如果最长 task 的耗时是中间值的好几倍基本可以确认是倾斜。处理倾斜的手段分两类一类是改变分区策略比如用RangePartitioner替代哈希分区或者对热点 key 加随机前缀做两阶段聚合另一类是给倾斜的 task 更多资源通过spark.locality.wait和推测执行来缓解但这是治标。推测执行值得单独提一句。它会在检测到某个 task 明显慢于同批次其他 task 时启动一个副本同时跑谁先完成用谁的结果。源码里由TaskSetManager的checkSpeculatableTasks负责判断阈值由spark.speculation.multiplier和spark.speculation.quantile控制。开启它能缓解倾斜带来的长尾但会增加资源消耗在资源紧张的集群上要谨慎。6.4 常见问题速查表现象源码层面的可能原因排查动作处理建议任务卡在 99% 长时间不动少数 task 处理数据量过大stage 需等全部完成看 Spark UI 的 task 耗时分布处理数据倾斜开推测执行同一个 RDD 被计算多次未缓存每次 action 都重走compute链检查storageLevel是否为 NONE对复用节点 persistcheckpoint 后耗时翻倍checkpoint 触发额外物化 job上游未缓存看是否只有一个 action先 persist 再 checkpoint报分区下标不连续异常getPartitions返回值中index与数组下标不一致检查自定义 RDD 的分区构造保证 index 从 0 连续递增executor 端空指针闭包里引用了SparkContext或 driver 侧对象看堆栈是否在compute附近移除闭包中的 driver 引用序列化失败闭包捕获了不可序列化对象报错信息会指出字段名用transient或 broadcastshuffle fetch 失败率高单次拉取过大或网络抖动看 executor 日志的 retry 记录调小maxSizeInFlight增加重试分区数远大于预期小文件过多或minPartitions未生效看输入文件数量和大小预处理合并小文件内存一直涨不降缓存未释放或存储级别为反序列化对象检查是否有unpersist统一管理缓存生命周期join 走了 shuffle 但本可避免两侧分区器和分区数不一致打印两侧partitioner对齐分区策略这张表是我这几年攒下来的基本覆盖了日常八成的问题现象。用法是先按现象匹配再看源码层面的原因最后按排查动作确认。别跳过确认直接改参数我见过太多人一看到 OOM 就加内存结果根因是数据倾斜加了内存只是把爆炸时间往后推。最后分享一个我自己的习惯每次遇到需要翻源码才能解释的问题我都会在本地留一个最小复现的 demo把相关 RDD 的partitions.length、partitioner、dependencies打印出来。这几个信息一摆绝大多数“为什么慢”“为什么重算”“为什么切了这么多 stage”的问题答案就写在眼前了。源码不是用来膜拜的是用来当放大镜的你越早把它当成日常工具调优这件事就越少靠猜。