DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行

上一篇解决了 Stage 的划分规则——遇到 ShuffleDependency 就切一刀。这一篇进入调度器内部。

Stage 划分出来之后，谁来决定 Stage 的提交顺序？谁来把 Stage 拆成 Task 发给 Executor？Spark 用两层调度器分工完成这件事：DAGScheduler 负责 Stage 级别的依赖分析和提交顺序，TaskScheduler 负责 Task 级别的资源分配和执行调度。

本文只抓一个问题：一个 action 触发之后，从 Job 到 Stage 到 Task 再到 Executor，调度链路上每一步发生了什么。

图 1：Driver 内部的调度组件与 Executor 的执行边界

flowchart TB
  A[action] --> D[DAGScheduler：划分 Stage 并生成 TaskSet]
  D --> T[TaskScheduler：选择 Task 与 Executor]
  T --> B[SchedulerBackend：对接执行资源]
  B --> E[Executor：运行 Task]
  E -->|完成或失败事件| D

图 1 中 DAGScheduler 和 TaskScheduler 都属于 Driver。集群管理器启动 Executor 之后，仍由 Spark 内部调度器决定哪些分区先执行；TaskSet 只需包含本次缺失或待计算的分区。

调度全景

用户代码: rdd.count()
      │
      ▼
  SparkContext.runJob()
      │
      ▼
  DAGScheduler.submitJob()
      │ 构建 Stage DAG，按依赖顺序提交
      ▼
  DAGScheduler.submitStage()
      │ 检查父 Stage 是否完成
      ▼
  DAGScheduler.submitMissingTasks()
      │ 把 Stage 拆成 TaskSet
      ▼
  TaskScheduler.submitTasks(TaskSet)
      │ 把 Task 分配给 Executor
      ▼
  SchedulerBackend.launchTask()
      │ 序列化 Task，通过 RPC 发送
      ▼
  Executor.launchTask()
      │ 反序列化，执行 Task.run()
      ▼
  结果返回 Driver

这条链路分成两个清晰的职责边界：DAGScheduler 只关心"哪些 Stage 先跑、哪些后跑"，TaskScheduler 只关心"这个 Task 放到哪个 Executor 上跑"。

DAGScheduler：Stage 依赖与提交顺序

DAGScheduler 的核心工作是把 RDD 依赖图转化成 Stage DAG，然后按拓扑序提交。

当一个 action 触发 runJob 时，DAGScheduler 从最终 RDD 开始，递归回溯依赖链，构建出完整的 Stage DAG。回溯规则是上一篇介绍的：窄依赖合并到当前 Stage，宽依赖切出新的 ShuffleMapStage。

Stage 之间的提交顺序遵循一条简单规则：一个 Stage 只有在它的所有父 Stage 都完成后才能提交。DAGScheduler 维护一个等待队列（waitingStages），当一个 Stage 的所有 Shuffle 输出都就绪后，把它移到运行队列。

Stage DAG 示例（wordcount + sort）：

  Stage 0: textFile → flatMap → map
      │ (Shuffle)
      ▼
  Stage 1: reduceByKey
      │ (Shuffle)
      ▼
  Stage 2: sortByKey → collect

提交顺序：Stage 0 → Stage 1 → Stage 2

Stage 0 没有父 Stage，可以立即提交。Stage 1 依赖 Stage 0 的 Shuffle 输出，必须等 Stage 0 的所有 Task 完成。Stage 2 依赖 Stage 1 的 Shuffle 输出，必须等 Stage 1 完成。

如果 DAG 中有独立的分支（比如两个不相关的 RDD 各自做 reduceByKey 后再 join），独立分支的 Stage 可以并行提交。

TaskScheduler：资源分配与数据本地性

当 DAGScheduler 决定提交一个 Stage 时，它把 Stage 中的所有分区封装成一个 TaskSet，交给 TaskScheduler。TaskSet 包含一组 Task，每个 Task 对应 Stage 中的一个分区。

TaskScheduler 的职责是把这些 Task 分配给集群中的 Executor。分配策略的核心考量是数据本地性——尽量把 Task 调度到数据所在的节点，减少网络传输。

TaskScheduler 为每个 Task 维护一组首选位置（来自 RDD 的 preferredLocations），然后按优先级尝试调度：

调度优先级（从高到低）：

PROCESS_LOCAL  数据在同一 Executor 内存中         等待 3s
NODE_LOCAL     数据在同一节点的另一个 Executor 中   等待 3s
RACK_LOCAL     数据在同一机架的另一台机器上          等待 3s
ANY            数据在任意位置                      立即调度

如果最优级别的 Executor 暂时没有空闲 CPU 核心，TaskScheduler 会等待一段时间（默认 spark.locality.wait=3s），超时后降级到下一个级别。这个等待时间是一个权衡：等太久浪费时间，不等又增加网络 I/O。

ShuffleMapTask 与 ResultTask

Stage 中的 Task 分为两种类型：

ShuffleMapTask：属于 ShuffleMapStage，任务是计算分区数据并写入 Shuffle 文件。每个 ShuffleMapTask 的输出是一组 Shuffle 文件（按下游分区号分桶），存储在本地磁盘上。

ResultTask：属于 ResultStage（最终 Stage），任务是计算分区数据并把结果返回给 Driver。collect 操作的每个 ResultTask 返回一个分区的全部数据；count 操作的每个 ResultTask 返回一个数字。

Stage 0 (ShuffleMapStage):
  ShuffleMapTask 0 → 写 Shuffle 文件
  ShuffleMapTask 1 → 写 Shuffle 文件
  ShuffleMapTask 2 → 写 Shuffle 文件

Stage 1 (ResultStage):
  ResultTask 0 → 返回结果给 Driver
  ResultTask 1 → 返回结果给 Driver

ShuffleMapTask 完成后，不直接把数据发给下游，而是向 Driver 报告 Shuffle 文件的位置（MapStatus）。下游 Stage 的 Task 启动时，从 MapOutputTracker 查询这些位置，然后从对应的 Executor 上拉取 Shuffle 数据。

实验：在 Spark UI 中追踪调度链路

在 spark-shell 中执行一个包含两次 Shuffle 的作业：

val data = sc.textFile("README.md")
val words = data.flatMap(_.split(" ")).map(w => (w, 1))
val counts = words.reduceByKey(_ + _)       // Shuffle 1
val sorted = counts.sortByKey()              // Shuffle 2
sorted.collect()

打开 Spark UI（http://localhost:4040），观察以下信息：

Jobs 页面：检查排序结果对应的 Job；sortByKey 为构建 RangePartitioner 可能先触发采样等额外 Job，因此不能预先断言整个示例只产生一个 Job。

Stages 页面：

• Stage 0：ShuffleMapStage，对应 textFile → flatMap → map。Shuffle Write 列显示写出的数据量。
• Stage 1：ShuffleMapStage，对应 reduceByKey。Shuffle Read 列显示从 Stage 0 拉取的数据量，Shuffle Write 列显示写出的排序前数据。
• Stage 2：ResultStage，对应 sortByKey → collect。Shuffle Read 列显示从 Stage 1 拉取的数据量。

点击某个 Stage 进入详情页，可以看到每个 Task 的执行信息：

• Locality Level：显示该 Task 实际使用的数据本地性级别。
• Duration：执行耗时。
• Shuffle Read/Write：该 Task 读写的 Shuffle 数据量。
• GC Time：垃圾回收耗时。

Task 失败与重试

Task 执行可能失败（OOM、节点故障、网络超时）。TaskScheduler 的重试策略：

spark.task.maxFailures=4 表示同一个 Task 连续失败 4 次后放弃，通常对应首次尝试加最多 3 次重试。前几次重试会把 Task 调度到不同的 Executor 上，避免反复在同一个坏节点上失败。

如果一个 ShuffleMapTask 失败，只影响该 Task 对应的分区。其他分区的 Shuffle 输出不受影响。

如果一个 Executor 完全失去响应，TaskScheduler 会把该 Executor 上所有正在运行的 Task 标记为失败，重新调度到其他 Executor。对于 ShuffleMapTask，若输出无法再从 ESS、保留的存储或其他可用路径读取，之前成功的 ShuffleMapTask 也可能需要重跑；Executor 进程退出不必然等于节点磁盘内容丢失。

不同 Task 的失败次数不会简单相加后触发 spark.task.maxFailures。Stage 连续尝试还受 spark.stage.maxConsecutiveAttempts 等独立阈值约束。

模式提炼

Spark 的两层调度器体现了职责分离模式：逻辑调度和物理调度解耦。

DAGScheduler 处理逻辑层面的问题——Stage 之间的依赖关系和提交顺序。它不关心集群有多少台机器、每台机器有多少 CPU。TaskScheduler 处理物理层面的问题——每个 Task 放到哪个 Executor 上、数据本地性如何权衡。它不关心 Stage 之间的依赖关系。

这种分离让 Spark 可以用不同的 TaskScheduler 实现对接不同的集群管理器（YARN、Kubernetes、Standalone），而 DAGScheduler 的逻辑完全不变。

同样的模式出现在操作系统的进程调度中：高层调度器（Job Scheduler）决定哪些作业进入就绪队列，低层调度器（CPU Scheduler）决定哪个就绪进程获得 CPU 时间片。

工程迁移表

Spark 概念	MapReduce 对应	Flink 对应	数据库对应
DAGScheduler	JobTracker（MR1）/ ApplicationMaster（MR2）	JobManager 的 ExecutionGraph 调度	查询协调器（Query Coordinator）
TaskScheduler	TaskTracker（MR1）/ Container 分配	TaskManager 的 Slot 管理	执行器的线程池调度
Job	MapReduce Job	Flink Job	一次查询执行
Stage	Map 阶段 / Reduce 阶段	Operator Chain（ExecutionVertex）	查询计划的 Fragment
TaskSet	Map Task 集合 / Reduce Task 集合	同一 Chain 的并行 SubTask	Fragment 的并行实例集合
数据本地性调度	InputSplit 的 location hint	InputSplitAssigner	分区亲和性
Task 重试	speculative execution + 重试	checkpoint 恢复	查询重试

常见误解

误解一：“Stage 之间一定是串行执行”。如果 DAG 中有多个独立分支（没有依赖关系的 Stage），这些 Stage 可以并行执行。只有存在依赖关系的 Stage 才需要串行。

误解二：“Task 数等于 CPU 核心数”。Task 数由 Stage 中的分区数决定，与 CPU 核心数无关。如果 Task 数多于可用核心数，多余的 Task 排队等待。如果 Task 数少于核心数，部分核心空闲。

误解三：“DAGScheduler 是一个独立进程”。DAGScheduler 运行在 Driver 进程内部，是 SparkContext 的一个组件。它通过事件队列（DAGSchedulerEventProcessLoop）异步处理 Stage 提交、Task 完成等事件。

练习

1. 构造一个 DAG 包含两个独立分支（两个不相关的 reduceByKey），最后 join 在一起。在 Spark UI 中观察两个分支的 Stage 是否并行执行。

2. 修改 spark.locality.wait 参数（设为 0 和 10s），对同一个读 HDFS 文件的作业执行两次，对比 Task 的 Locality Level 分布。思考等待时间对总执行时间的影响。

3. 在 Spark UI 的某个 Stage 详情中，找出执行时间最长的 Task 和最短的 Task。计算两者的比值。如果比值超过 10 倍，可能存在什么问题？

系列导航

序号	主题	状态
00	导读：为什么 Spark 的核心是一张 DAG	已完成
01	RDD：弹性分布式数据集的五大属性	已完成
02	宽依赖与窄依赖：Stage 是怎样划分出来的	已完成
03	DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行	本篇
04	Shuffle 机制：数据跨分区交换的代价与优化	下一篇
05	内存管理与 Tungsten：堆外内存、序列化与代码生成
06	存储体系：Block Manager、广播变量与累加器
07	容错机制：Lineage、Checkpoint 与推测执行
08	Spark SQL 与 Catalyst：从 SQL 文本到物理算子树
09	DataFrame 与 Dataset：类型安全与性能的折中
10	Adaptive Query Execution：运行时改写执行计划
11	数据源 API 与谓词下推：让存储层帮忙过滤
12	Structured Streaming：把流当成无界表
13	水位线与状态管理：流处理的时间语义
14	Spark on YARN / K8s：资源管理与部署模式
15	动态资源分配与作业调度：多租户集群的资源博弈
16	性能诊断：Spark UI、Event Log 与调优思路
17	Spark 与 Flink：两种计算引擎的设计选择

参考资料

• Apache Spark 源码：core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala https://github.com/apache/spark
• Apache Spark 源码：core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
• Apache Spark 官方文档：Cluster Mode Overview. https://spark.apache.org/docs/latest/cluster-overview.html
• Matei Zaharia. An Architecture for Fast and General Data Processing on Large Clusters. PhD Thesis, UC Berkeley, 2014. Chapter 4.