2. Spark作业运行机制深度解析：DAG、Stage、Task的划分逻辑
好，咱们今天聊聊Spark作业到底是怎么跑起来的。很多同学写Spark代码时，只管调API，跑通了就完事。但一旦遇到性能问题，就抓瞎了。为什么？因为你不懂底层是怎么划分任务的。
我个人习惯，拿到一个Spark作业，第一件事就是看它的DAG图。说白了，DAG就是Spark作业的执行蓝图。你想想看，没有这张图，Spark怎么知道先算哪个、后算哪个？
2.1 从代码到DAG：你的逻辑是怎么被翻译的？
你写一段Spark代码，比如：
val rdd = sc.textFile("hdfs://data/input")
val words = rdd.flatMap(_.split(" "))
val pairs = words.map(word => (word, 1))
val counts = pairs.reduceByKey(_ + _)
counts.saveAsTextFile("hdfs://data/output")
这段代码在Spark眼里，其实是一连串的依赖关系。每个RDD都记录了它是从哪来的、经过了什么变换。Spark会把这些RDD的依赖关系串起来，形成一张有向无环图——也就是DAG。
核心要点：DAG记录了RDD之间的血统关系。窄依赖（如map、filter）是1对1的，宽依赖（如reduceByKey、groupBy）是1对多的。这个区别，直接决定了Stage怎么切分。
我在项目中遇到过一位同事，他写了个超长的链式调用，一个RDD上连续做了十几个map和filter。结果DAG图长得吓人，但性能反而没提升。为什么？因为窄依赖再多，也不会触发Stage切分，但每个步骤都要序列化数据，反而增加了开销。
2.2 Stage划分：宽依赖就是分水岭
DAG有了，接下来Spark要决定怎么切分Stage。这里有个关键原则：遇到宽依赖就切一刀。
为什么会这样？你想想看，窄依赖（比如map）每个分区独立计算，不需要等别的分区数据。但宽依赖（比如reduceByKey）需要把所有分区中相同key的数据拉到一起，这就产生了Shuffle。Shuffle意味着数据要跨网络传输，这是一个昂贵的操作。
Spark的DAGScheduler会从后往前遍历DAG，遇到宽依赖就切分Stage。每个Stage内部全是窄依赖，可以流水线式执行。Stage之间则通过Shuffle连接。
| 依赖类型 | 特点 | Stage划分 | 典型算子 |
|---|---|---|---|
| 窄依赖 | 每个父分区只被子分区使用一次 | 同一Stage内 | map、filter、union |
| 宽依赖 | 每个父分区被子分区多次使用 | 触发新Stage | reduceByKey、groupBy、join |
我的经验：调优时，我通常会先数一数作业有几个Stage。Stage太多，说明Shuffle太多，网络开销大。我曾经优化过一个作业，把3次reduceByKey合并成1次，Stage从4个降到2个，执行时间缩短了60%。
2.3 Task划分：Stage里的最小执行单元
Stage划分好了，接下来就是Task。每个Stage会被拆成多个Task，每个Task负责处理一个分区（Partition）的数据。
这里有个容易混淆的点：Task的数量 = 分区数。不是由你写的算子数量决定的，而是由RDD的分区数决定的。
举个例子，你从HDFS读文件，默认每个Block（128MB）对应一个分区。如果文件有100个Block，那第一个Stage就有100个Task。每个Task处理一个分区的数据，执行相同的计算逻辑（比如map、filter）。
注意：Task不是越多越好。我见过有人把分区设成10000个，结果每个Task处理的数据量极小，但调度开销巨大。反过来，分区太少也不行，资源利用不充分。一般建议每个Task处理100MB-1GB的数据，或者每个CPU核心分配2-4个Task。
2.4 完整流程：从提交到执行
嗯，咱们把整个流程串起来捋一遍：
- 提交作业：你调用action算子（如saveAsTextFile、collect），触发作业提交。
- 构建DAG：DAGScheduler根据RDD的依赖关系，构建DAG图。
- 划分Stage：从后往前遍历DAG，遇到宽依赖就切分Stage。
- 生成TaskSet：每个Stage被拆成多个Task，每个Task对应一个分区。
- 提交Task：TaskScheduler将Task分配到Executor上执行。
- 执行与重试：Task执行失败会自动重试（默认3次）。
我记得有一次线上作业频繁失败，查了半天发现是某个Executor内存不足，导致Task反复OOM。后来调整了分区数，让每个Task处理的数据量小一些，问题就解决了。你看，不懂Task划分逻辑，连问题都定位不准。
2.5 避坑指南：我踩过的那些坑
我曾经犯过一个低级错误：在reduceByKey之后又跟了一个map，以为map是窄依赖不会影响Stage。但reduceByKey已经触发了Shuffle，生成了新的Stage。那个map其实是在新Stage里执行的，并没有减少Stage数量。
还有一次，我为了图方便，在循环里反复调用groupByKey。结果DAG图里Stage数量爆炸，每个Stage都要做一次Shuffle。后来改成用aggregateByKey，一次Shuffle搞定所有聚合操作。
总结一下：
- DAG是蓝图，Stage是阶段，Task是执行单元。
- 宽依赖是Stage的分水岭，尽量减少Shuffle次数。
- Task数量由分区数决定，不是算子数量。
- 调优时，先看DAG图，数Stage，再调分区。
好了，这一章就到这里。下一章咱们聊聊资源管理，怎么给Spark作业分配CPU和内存，才能跑得又快又稳。