Spark DAG执行引擎详解：从RDD血缘到Stage调度与生产调优

引言：为什么要专门聊Spark的DAG

“开源组件系列”写到这里，Spark终于出场了。十多年的大数据开发生涯里，我用过的计算引擎不少，从最早期的MapReduce，到后来的Tez、Flink，再到Spark，如果非要选一个“最值得搞懂计算引擎本质”的组件，我会把票投给Spark，更准确地说，投给它的DAG执行引擎。

这个问题先想明白：为什么大家都在说“Spark快”？很多人第一反应是“内存计算”。这个说法对，但不全对。MapReduce时代也有人把数据放内存，没见谁把MR调出Flink的延迟。Spark真正拉开差距的地方，是它把一个复杂的计算任务打散成一个有向无环图，也就是DAG，然后由调度引擎自动拆解、并行化、容错重试。这套机制让Spark既能跑批处理，也能扛流式计算，还能做机器学习迭代——一鱼多吃，靠的就是DAG这张“计算蓝图”。

这篇文章不打算抄官方文档，我尽量用做项目的视角，把DAG的执行链路从用户代码到Executor上的任务跑完，讲清楚每个环节为什么这么设计、实际生产里会遇到哪些坑。无论你是刚接触Spark的新手，还是已经被数据倾斜折磨过几轮的老人，读完应该都能对“Spark任务到底是怎么跑起来的”有一个完整画面。

1. 内容整体设计与思路拆解

1.1 为什么是DAG而不是别的

要理解Spark的设计选择，得先看它拒绝了什么。

早期的MapReduce实际上是一个“两阶段固定模型”：Map结束落盘，Reduce拉数据再算。所有逻辑必须硬塞进这两个阶段里。遇到一个需要三次关联、两次分组、一次排序的复杂分析，MR的做法是用一个Job链串起来，每个Job之间通过HDFS中间结果衔接。结果就是：同一个数据被反复读写磁盘，几十个MR串起来跑一个报表成了家常便饭。

DAG的思路完全不同。它允许你把整个计算逻辑表达为一张图，图的每个节点是一次计算（RDD或DataFrame的转换），边是数据流向。这张图不需要人为切成一个个Job，而是由引擎自动做“调度切分”——把那些必须串行的环节（Shuffle）作为分界线，切出一个个Stage，Stage内部再最大限度地并行跑。中间过程能留内存就留内存，能不落盘就不落盘。

这个设计带来的第一个好处是：计算模型不再扭曲。业务逻辑怎么表达，DAG就怎么画，人脑和引擎之间的翻译成本降到最低。第二个好处是：执行计划可以全局优化。因为整张图在面前，引擎可以在切Stage之前做谓词下推、列裁剪、算子合并，这些东西在“多个MR Job串联”的模式下几乎无从谈起。

第三个好处，也是最容易被忽视的：容错代价大幅下降。MR的容错粒度是整个Job重跑，Spark的容错粒度是单个RDD分区。某个Executor挂了，只需要根据血缘把丢失的分区重新计算一遍，不用整个任务推倒重来。这个特性在跑长任务时简直就是救命稻草——我见过一个跑了6小时的ETL，中途挂掉一台机器，DAG自动补数据继续跑完，换成MR估计就得从头再来。

1.2 几个必须建立的核心概念

聊DAG之前，有几个概念不捋清楚，后文会看得很吃力。我用最直白的方式说一遍。

RDD（弹性分布式数据集）：不要把它想象成一个“数据集”，更准确地说，它是一个“计算指令集合”，里面存的是从哪读数据、经过哪些变换（map、filter、join），以及怎么计算分区的逻辑。数据本身不在RDD里，数据在RDD的compute方法被调用后才产生。这也是为什么RDD可以不存任何数据、却依然能把整个计算链串起来的原因。

血缘（Lineage）：每个RDD都记得自己的“爸爸是谁”（父RDD）以及自己是怎么从父RDD变出来的（算子）。血缘链就是一张不折不扣的DAG。有了血缘，某块数据丢了，计算引擎能从源头重新算回来，这就是“血统容错”的底层依据。

窄依赖与宽依赖：这是划分Stage的唯一标准。窄依赖的意思是父RDD每个分区最多被子RDD的一个分区使用，典型算子有map、filter、union；宽依赖则是父RDD的一个分区会被子RDD的多个分区使用，典型算子有groupByKey、join、distinct。窄依赖可以“原地手拉手”执行，不需要Shuffle；宽依赖必须全量数据洗牌重组，是Stage的天然分界线。

Stage与Task：每个RDD的每个分区对应一个Task，Task是真正跑到Executor上执行的最小单元。多个Task按依赖关系组成Stage，Stage之间串行，Stage内部并行。DAGScheduler的核心工作就是把这个Stage的边界画出来。

概念备齐了，接下来进入正题：一张用户代码里的dataframe处理逻辑，怎么一步步变成DAG，再变成一堆Task的。

2. 核心细节解析与实操要点

2.1 DAG的构建时机：Action才是发令枪

很多刚接触Spark的人都会误以为，写出val rdd = sc.textFile(...)之后，DAG就已经构建好了。实际完全不是这样。

DAG的构建发生在Action算子执行时。transformations（map、filter、join等）只负责“往树上挂新节点”，它们返回的还是抽象表示，什么计算都没发生。只有Action（count、collect、saveAsTextFile等）出现，才会触发一个Job的提交，沿着血缘链向前回溯，把整棵RDD依赖树“冻结”成一张DAG。

这个设计的意义在于懒加载与复用。同一份原始数据，先过滤得到A，再分组得到B，只要你最后不调用Action，这些计算都不消耗资源；等你调了两次Action，引擎还有机会把两条血缘链的公共部分缓存起来，避免重复计算。用一句话概括：transformations描述“要做什么”，Action才决定“做一次”。

实际生产中，这个机制带来的最常见问题就是Action过多导致重复计算。我见过一个案例，某同学一口气在同一个DataFrame上连续调用了三次count/collect，每次count都会触发一遍完整的前置计算。当时的解决方式很简单：在第一次Action之前用cache/persist把关键DataFrame固定住，后续Action直接读缓存，成本从三次全量计算降为一次。

2.2 Stage的切分：找到那堵墙

DAGScheduler从Action对应的RDD出发，向前遍历血缘链，每遇到一个宽依赖，就切一刀。宽依赖的算子（reduceByKey、join、groupBy、distinct等）背后都跟着一次Shuffle，Shuffle意味着数据要在节点之间重新分配，这是无法绕过物理边界的“计算墙”。墙内，Stage之间串行执行；墙外，各Stage内部可以随便并行。

一个典型的shuffle依赖流程是：Stage0读取HDFS数据，做map、filter等窄依赖操作，然后在一个reduceByKey处画上“闸门”，切开Stage0和Stage1。Stage0的Task把结果按Key做哈希分区并写入本地磁盘（Shuffle Write），Stage1再去拉取（Shuffle Read）。整个过程，Shuffle既是“数据交换机制”，又是“Stage划分依据”。

这里面有一个细节值得展开：同一个算子，在不同数据分布下，依赖类型可能不同。join就是典型。如果两边数据集都已按Join Key做了分区与排序，引擎可以走Broadcast Hash Join或者SortMerge Join，避免大规模Shuffle；如果有一方足够小，还可以优化成Map端广播Join。DAGScheduler在划分Stage时，是“实际发生在父子RDD分区之间的依赖”为准，而不是算子类型本身。理解了这点，你就明白为什么Spark的物理计划和逻辑计划不是一回事，也明白日常调优的重点其实大多集中在“如何减少Stage数量、避免多余的墙”。

2.3 一个DAG执行计划的完整链路

用一张实际的执行计划来串一遍。假设有一段这样的代码：

val raw = spark.read.parquet("/data/events") val filtered = raw.filter(col("event_type") === "click") val grouped = filtered.repartition(200, col("user_id")) .groupBy("user_id").agg(count("click_count")) val result = grouped.sort(col("count").desc) result.write.mode("overwrite").parquet("/data/result")

这段逻辑翻译成DAG是这样的链路：

• read.parquet：创建底层的FileSourceRDD，每个分区对应一个Parquet文件块。
• filter：窄依赖，直接挂在父RDD上，形成Stage0的第一组操作。
• repartition：显式打散数据，产生宽依赖，Stage0在这里结束，Stage1从Shuffle Read开始。
• groupBy/agg：groupBy换一个角度理解，本质上是“按key重新分组 + 对同组数据做聚合”。聚合可以在两个阶段完成：map端先做部分聚合，reduce端再做全量聚合。这个过程也需要Shuffle，因此在Stage1和Stage2之间切一刀。
• sort：排序在聚合结束后，在同一个Stage底下完成即可。
• write：由DataFrameWriter触发，这既是Action，也是最终的DAG出口。

所以整张DAG共三个Stage，Stage0负责读+过滤+分区，Stage1负责Shuffle聚合，Stage2负责排序写最终结果。实际用explain打印执行计划时，你会发现树形结构比这精细得多，还会多出Exchange（Shuffle）、Sort等物理算子。但骨架就是这三个Stage，这已经足够说明问题了。

我建议每个Spark用户都养成一个习惯：跑任务之前，先打印一遍物理计划。explain不花资源，却能把Stage边界、Shuffle位置、Join策略一览无余。哪些环节走了不必要的Shuffle、哪些Join没用上广播，一照妖就现形。

3. 实操过程与核心环节实现

3.1 调度内核：DAGScheduler与TaskScheduler各干各的活

DAG构建好之后，真正把它“跑起来”的是Spark的调度内核。这个内核由两个明显分工的组件构成。

DAGScheduler是高层的“总导演”。它接到一个Job（由一个Action触发），就执行四件事：一是构建DAG；二是切分Stage，并且记录每个Stage的Shuffle依赖信息；三是为每个Stage生成一组Task，每个分区对应一个Task；四是提交TaskSet给下层调度器。同时，DAGScheduler还负责处理Shuffle输出丢失、任务失败导致的Stage重试。它的眼睛只看“图”的层面，不和Executor直接打交道。

TaskScheduler是低层的“现场调度员”。它从DAGScheduler那里接到TaskSet，按资源情况把Task分配到具体的Executor上。它会维护每个Executor的资源余量、跟踪每个Task的状态、处理Task失败重试（默认重试4次），并且维护“推测执行”机制——如果某个Task跑得明显比同Stage的其他Task慢，调度器会复制一份同样的Task到别的节点去跑，谁先结束用谁的结果。

两个调度器之间的交互协议很简单：DAGScheduler提交TaskSet，TaskScheduler返回Task完成/失败的信号，通知驱动端更新Job状态。这个“高层不掺和低层细节、低层不关心全局图”的分层设计，是Spark可以无缝接入Yarn、Mesos、Kubernetes等多种集群管理器的关键——DAG的图构建逻辑不依赖具体部署环境，资源适配全部下沉到TaskScheduler。

实际调试时，Spark UI的“Jobs”和“Stages”页面其实就映射了这两个层：Jobs页展示的是DAGScheduler视角的Action与Job状态，Stages页展示的是Stage内的Task分布与执行情况。看UI时先盯这两个Tab，问题定位会快很多。

3.2 Executor上的任务执行：一场微观接力赛

Task被分配到Executor之后，执行的微观流程大致如下：

1. Executor收到Task的序列化二进制（Task二进制里包含了RDD的compute逻辑与闭包依赖）。
2. Executor反序列化Task，创建TaskRunner线程，加入线程池调度，按Executor核数并行执行。
3. TaskRunner调用RDD的compute方法，开始读取Shuffle输出或数据源分区。
4. RDD算子链逐层执行，每层发生一次传递，直到得到该分区最终结果。
5. 如果Task产出的是Shuffle数据，写入本地磁盘的Shuffle文件，同时更新Driver端的BlockManager元数据，记录“哪个Stage的哪个分区在哪台机器上”。
6. 如果Task产出的是最终结果（如collect的返回值），则通过网络回传Driver。

这个流程里，最容易理解错的地方是“RDD链条计算本身是分层的，但物理执行是单线程串行的”。什么意思呢？假设一个分区要经过map → filter → flatMap三层变换，物理上不运行三个独立的map/flatMap过程，而是由一个Task在一个循环里依次调用三次函数映射。这也是Spark比MR少了大量序列化/反序列化开销的原因之一——中间过程没有落盘，没有跨进程的数据传递。

写代码的时候心里默念这句话：一个Task只处理一个分区，完整跑完分区的转换链，计算出最终结果才提交。所以：

• 每个Task能处理的“逻辑复杂度”由血缘链深度决定，链越长，单个Task的CPU开销越大。
• 分区数量决定Task总数，进而决定Spark能并行到什么程度。分区太少，CPU闲着；分区太多，调度和序列化开销拖后腿。经验法则：每个分区处理数据量控制在128MB左右，对应HDFS默认块大小。

3.3 两个生产环境的提交与验证案例

案例一：用Yarn模式提交一个ETL任务并验证Stage划分

spark-submit \ --master yarn \ --deploy-mode cluster \ --driver-memory 4g \ --executor-memory 8g \ --executor-cores 4 \ --num-executors 20 \ --conf spark.sql.shuffle.partitions=200 \ --conf spark.shuffle.file.buffer=64k \ --class com.example.etl.processEventLog \ event-etl.jar

提交之后第一件事：打开Spark UI，进“Stages”页，观察整个Job被切成几个Stage，每个Stage的Duration、Shuffle Read/Write体积。如果是三四个Stage且每个Stage的Task执行时间都很均匀，那么这个提交基本健康。如果出现某个Stage只有两三个Task但耗时是所有其他Stage之和，那基本可以断定数据倾斜来了。

案例二：流式计算任务的DAG表现

val stream = spark.readStream.format("kafka") .options(Map("kafka.bootstrap.servers" -> "...", "subscribe" -> "events")) .load() val agg = stream.groupBy("user_id").count() agg.writeStream.outputMode("update") .format("console") .trigger(Trigger.ProcessingTime("10 seconds")) .start()

Structured Streaming跑起来之后，DAG图上会出现一个有意思的结构：一个批次的处理仍然是“读Kafka → 聚合 → 输出”的DAG，但在StreamingQuery内部还有一个“跨批次的状态管理”环节，StateStore负责保存和更新用户累计计数。这意味着每个批次DAG的Stage0都是读取最新的Kafka偏移量，Stage1把新数据与历史State做合并，再输出。使用状态存储的流任务，对StateStore的Compact策略要特别关注，它会定期把增量日志合并成快照，这个操作如果被配置成同步执行，会出现偶发的延迟尖刺。

说实话，先看DAG再调优，这是唯一正经的路径。上来就调spark.executor.memory的人，多半会在某个深夜被OOM打脸。

4. 常见问题与排查技巧实录

4.1 最常见坑之一：数据倾斜在DAG层面的表现

数据倾斜是Spark生产环境里出现频率最高的性能问题，在DAG图上有着非常典型的“长相”。

正常情况下，一个Stage内的Task执行时间应该集中在均值附近，像一排房子门头差不多高。一旦发生倾斜，DAG图上就会出现“一根柱子捅破天际”的画面——一个或者少数几个Task的执行时间呈指数级超出其他Task。

案例：某广告流量统计任务，join两份数据：一份是流量明细，一份是广告位维度表。维度表里有一个“默认广告位”的ID覆盖了全国六七成的流量，结果是reduce端聚合时，这一条Key把一大堆数据都吸到了同一个分区上，那个Task处理的数据量是其他Task的一千倍。

DAG层面的处理思路，围绕“把倾斜的Key拆开”展开：

• 加盐：把倾斜Key拼上随机后缀，打散到多个分区分别聚合，再按真实Key二次聚合。
• 广播：如果小表本身足够小（不超过Executor内存），用Broadcast Hash Join消除Shuffle，从根上消掉那堵“墙”。
• 过滤：倾斜Key如果本身对最终结果无意义，直接先行过滤。
• 两阶段聚合：先分区局部聚合，再全局聚合，最常见的SQL改写方式。

实战中我习惯先用spark UI + 日志确认倾斜Key具体是谁，再决定用加盐还是广播。上来就改代码而不先定位Key的，很可能白折腾一场。

4.2 常见坑之二：Shuffle文件拉取失败与Executor心跳丢失

Task跑完、Shuffle数据写完了，下一个Stage去其他节点拉数据的时候报了FetchFailedException，这是DAG执行里最让人头大的一类问题，因为它看起来不像代码问题，而是节点问题。

根据我的经验，大部分FetchFailed的根因是下面这几类：

• Executor内存不够，GC狂转：数据在BlockManager内存里被回收，拉取方找不到了。解决方向是调大Executor内存，或者开启spark.shuffle.detectCorrupt=false（不推荐，治标不治本）。
• 磁盘问题：Shuffle文件写到本地磁盘的途中遇到过节点磁盘SSD故障，文件写坏。检查节点磁盘IO和dmesg日志。
• 连接数被打满：大规模Stage，Reducer特别多，同时向一个节点发起Fetch，打到TCP连接上限。方向是调大spark.shuffle.io.maxRetries与spark.shuffle.io.retryWait，同时检查spark.shuffle.io.numConnectionsPerPeer。

注意一个关键点：FetchFailed并不一定导致整个Stage失败。DAGScheduler会把这个失败的Shuffle输出标记为“不可用”，然后尝试重新调度该Stage的上游——如果恰好某个上游Stage的分区本来就有备份（比如RDD本身被缓存了），引擎甚至可以只重算缺少的部分。这个容错粒度的差异，正是DAG引擎区别于MR的核心价值。当你看到UI里Stage自动重试又自动成功时，实际上就是血统容错在工作。

4.3 常见坑之三：OOM并没那么容易定位

Spark任务的OOM在DAG图上经常“伪装”成两种情况：一种出现在某个Task内部，日志报Java heap space；另一种是整个Executor挂掉，出现Container killed by YARN for exceeding memory limits。

第一种情况其实更好排查——通常是单个分区里攒了太多数据量。比如在groupBy之后没做局部聚合直接collect到Driver；或者在executor里cache了一个特别大的DataFrame。解决办法是控制分区粒度、用持久化级别MEMORY_AND_DISK而不是MEMORY_ONLY、以及避免大对象闭包传递。

第二种情况我踩过坑的那次印象特别深：某系统每次跑凌晨的任务必然Executor退出，白天任务都正常。排查日志发现，凌晨任务读到的数据量是白天的五十倍以上，因为上游是凌晨统一刷库。那个任务的Executor内存只按白天数据量配了，于是到了凌晨直接爆掉。经验是：Executor内存配置按峰值负载跑一遍之后确定，不要按平均值。

4.4 小问题速查：一次看完直接照做

4.5 一套可以复制的排查路径

遇到问题，建议大家固定走下面这套路径，能省大量踩坑时间。

第一，进Stages页，选定异常Stage，看Summary Metrics里的执行时间、GC时间、Shuffle体积。倾斜一定会在这些指标里有表现。

第二，看Event Timeline，一条条Task的时间轴里，耗时大头是Computing、GC还是Shuffle读。哪个环节占比高，问题就在哪个环节。

第三，看Executor页，重点观察Executor的GCTime和已用内存。GC占比超过20%基本就说明内存吃紧。

第四，翻Driver日志里的错误栈。定位到具体代码行，再倒回来看它对应的RDD依赖链，最后根据依赖链猜测是分区还是Shuffle还是数据源的问题。

这套路径虽然朴素，但几乎所有生产事故都能在第四步之前定位到八八九九。

5. 性能调优：从DAG的角度重新审视一切

5.1 如何从执行计划反推优化点

拿到一张执行计划，好比拿到一张作战地图。看执行计划的顺序，我是这么看的：

第一，数Stage数量。Stage数量如果远大于你的业务逻辑复杂度，说明Shuffle太多了。多余的Shuffle不仅带来网络和磁盘开销，还可能引入数据倾斜。一个常见的“多Stage”原因是没必要的groupBy。比如某段逻辑最后只需要一个布尔值判断Key存不存在，却先对全表做了一次groupBy().count()，触发了巨大的Shuffle和Reduce。改成map后直接distinct或者first，Stage瞬间少一个。

第二，看Join的物理实现。执行计划里如果出现SortMergeJoin，说明两边分区规模都不小，有Shuffle；如果出现BroadcastHashJoin，说明一边被执行器广播了，没有Shuffle。从调优角度，能Broadcast就Broadcast，小表join大表时这招尤其管用。

第三，看Filter和Project在计划树里的位置。如果Filter被下推到TableScan之上，说明谓词下推做得不错；如果Filter在Join之后才出现，意味着大量数据白白参与了Join，这可能急需优化SQL逻辑本身。

5.2 核心参数的工业级配置经验

关于Spark调优参数，网上的说法五花八门，但真正经过长时间生产环境检验的配置，其实是有迹可循的。下面是我自己在多个集群上被验证有效的经验值，仅供参考，别硬背，按业务量调。

特别要提醒的是：Spark 3.0之后自适应查询执行（AQE）必须在开启状态。有了AQE，简化操作的空间大得多，比如repartition手动优化都不太需要考虑了，引擎会自己根据中间结果的规模去动态调分区数。

5.3 两个真实的DAG优化案例

案例A：三阶段变一阶段

曾经接过一个数据处理任务，每天早上要跑30分钟，逻辑本身不复杂，就是对一张宽表做过滤后输出十几个客户明细。问题在于，处理是分三阶段：先抽全量数据到内存，再过滤出一份中间结果，再拿中间结果做join。每个阶段都独立执行，导致一份数据被读取三次、写两次中间表。

优化后的做法：所有逻辑一气呵成在一条DAG里完成，只在最终输出时刻写一次数据。同一份数据的读取次数由3次降为1次，总耗时从30分钟降到9分钟。这个案例最大的收获不是具体操作，而是在写Stage多、中间落盘多时，最好的优化往往是让动作在一条血缘链上完成，而不是拆开成多个Job。

案例B：AQE救场

某客户的离线报表任务，因为上游数据源在天级供应商分区数量不稳定，每天凌晨Spark生成的Task数量都在1万到5万之间波动。Task数少的时候，并发上不去；Task数多的时候，调度器和序列化开销显著增长。

在开启AQE并配上spark.sql.adaptive.coalescePartitions.enabled=true、spark.sql.adaptive.advisoryPartitionSizeInBytes=64m之后，Spark会在每个Shuffle Stage结束时自动把相邻小分区合并起来。最终任务执行时间稳定下来，长尾的、耗在“等待小任务完成”的时间近乎消失。梳理下来，AQE做的工作本质上就是：在DAG执行过程中，动态修改Stage内部的Task数量——把整张图的执行从“静态提交”变成“动态流水线”，这也是Spark 3.x在调度层面最大的进步。

5.4 写在最后的实用建议

关于DAG这一套，很多时候你已经知道理论正确与否，但生产环境的脏乱差会让人气急败坏。这里再分享我的几个建议，与其说参数，不如说是工作方法。

第一个建议：保持一个固定但合理的spark.sql.shuffle.partitions，不对每个任务临时改。稳定最重要。频繁按不同任务去微调分区数，会造成集群资源浪费，因为不同任务的分区数影响的是整个集群的调度效率。除非某个任务确实被检测到倾斜严重，否则全部不调。

第二个建议：务必保护Spark UI的运行历史。历史服务一旦关掉，事故复盘会非常痛苦，这是许多运维事故的隐藏根源。

第三个建议：把DAG图截图放进你的故障排查文档里。我自己的经验是，能画出问题的DAG图，比写一大段分析文字有用得多。这张图是“执行的真身”，而代码只是“逻辑的愿望”。

关于这个系列和下一步

我个人在实际操作中的体会是，理解Spark的DAG执行引擎，相当于拿到了Spark全部知识的“总钥匙”。读MR时代的东西可能需要一页页翻Job日志，读Spark则以一看DAG为分水岭——看得懂DAG，调优、查错、写复杂算子，都会一下子顺手很多。

这个系列里的下一篇文章，自然想写Spark SQL的执行引擎。如果说DAG是Spark的骨架，那Spark SQL Catalyst就是Spark的“翻译器和优化器”，它解决的是“同一张DAG，如何生成得更聪明”。之后再写Streaming和Structured Streaming时，DAG这套底子依然是贯穿始终的暗线。

在结束之前，有一个细节值得再啰嗦一句。有的同学在本地IDE写Spark程序时，习惯性地用一个“空Action”去debug血缘链，把DAG打出来看一眼就完事。这个动作本身很好，但别忽略一个问题：DAG的长相取决于触发Action的类型和前置数据源的实际分区数，在本地看和集群上看会有不小的差别。有条件的话，大任务一定上小规模集群数据先验一遍DAG再放大跑，这是我最想提醒的关键一条。