摘要：你是否清楚一个 Spark 应用中到底有几个 Job？为什么 .collect() 会触发 Job 而 .map() 不会？Application、Job、Stage、Task 之间的关系到底是什么？本文从四层执行层级全景图、Action 触发 Job 的源码链路、DAGScheduler 的 Stage 切分规则、多 Job 应用实战示例四个维度，配合 1 张原创深色架构图 + 完整源码分析，带你彻底理清 Spark 的执行层级体系。

关键词：Spark Application, Job, Stage, Task, Action, DAGScheduler, SparkContext, 执行层级

---

一、开篇：你写的一个 Application 到底有几个 Job？

先看一段代码，你能准确说出它会产生几个 Job 吗？

val lines = sc.textFile("hdfs:///data/words.txt")     // Transformation
val words = lines.flatMap(_.split(" "))                // Transformation
val pairs = words.map((_, 1))                          // Transformation
val counts = pairs.reduceByKey(_ + _)                  // Transformation

counts.collect()   // Action 1 → Job 0
counts.count()     // Action 2 → Job 1
counts.saveAsTextFile("hdfs:///output")  // Action 3 → Job 2

答案：3 个 Job。因为每个 Action 算子都会触发一个新的 Job。而 textFile、flatMap、map、reduceByKey 都是 Transformation——它们只是构建 DAG，不触发任何计算。

---

二、执行层级全景图

2.1 四层模型

Application (SparkContext)
  │
  ├── Job-0 (调用 collect() 触发)
  │     ├── Stage 0 (ShuffleMapStage: 2 Tasks)
  │     └── Stage 1 (ResultStage: 3 Tasks)
  │
  ├── Job-1 (调用 count() 触发)
  │     ├── Stage 2 (ShuffleMapStage: 2 Tasks)
  │     └── Stage 3 (ResultStage: 3 Tasks)
  │
  └── Job-2 (调用 saveAsTextFile() 触发)
        ├── Stage 4 (ShuffleMapStage: 2 Tasks)
        └── Stage 5 (ResultStage: 3 Tasks)

---

三、Action 触发 Job 的源码链路 🔥

// 源码：RDD.scala - collect()
def collect(): Array[T] = withScope {
  val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
  results.flatten
}

// 源码：RDD.scala - count()
def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum

// 源码：RDD.scala - saveAsTextFile()
def saveAsTextFile(path: String): Unit = {
  // ... 内部最终调用 sc.runJob()
}

// 核心链路
// Action → sc.runJob() → DAGScheduler.runJob()
//   → DAGScheduler.handleJobSubmitted()
//   → 创建 ActiveJob → Stage 切分 → submitStage()

3.1 哪些算子是 Action？

算子	返回类型	说明
collect()	Array[T]	拉取所有数据到 Driver
count()	Long	计数
take(n)	Array[T]	取前 n 个
reduce(f)	T	聚合
foreach(f)	Unit	遍历
saveAsTextFile()	Unit	保存到文件
first()	T	取第一个

反直觉点：reduceByKey 不是 Action！它是 Transformation，触发 Shuffle 但不触发 Job。

---

四、Stage 切分规则

// 源码：DAGScheduler.scala
private def getMissingParentStages(stage: Stage): List[Stage] = {
  stage.rdd.dependencies.flatMap {
    case shufDep: ShuffleDependency[_, _, _] =>
      // WideDep → 切分 → 创建父 ShuffleMapStage
      getOrCreateShuffleMapStage(shufDep, stage.firstJobId)
    case _ => Nil  // NarrowDep → 不切分，保持在同一 Stage
  }.toList
}

一句话规则：遇到 ShuffleDependency (Wide Dependency) 即切分 Stage。

---

五、多 Job 实战示例

val rdd = sc.parallelize(1 to 1000, 4)  // 4 Partitions

// Job 1: count
println(s"Count: ${rdd.count()}")       // Action → Job-1
// Job 2: collect
val arr = rdd.collect()                 // Action → Job-2 (无 Shuffle, 1 Stage)
// Job 3: save
rdd.saveAsTextFile("hdfs:///output")    // Action → Job-3

// 总计: 3 个 Job, 3×1=3 个 ResultStage

// 带 Shuffle 的场景
val rdd = sc.parallelize(1 to 1000, 4)
  .map(x => (x % 10, x))                // Narrow
  .groupByKey()                          // Wide! Stage 边界

rdd.count()  // Action → Job-1: Stage 0 (ShuffleMapStage) + Stage 1 (ResultStage)

---

六、Application/Job/Stage/Task 对比表

层级	定义	触发条件	数量
Application	SparkContext 实例	spark-submit	1
Job	一个 Action 的完整计算	Action 算子	1~N
Stage	Shuffle 边界切分的计算阶段	ShuffleDependency	每个 Job 1~N
Task	处理一个 Partition 的最小单元	Stage 内 Partition 数	每个 Stage 1~N

---

七、总结

要点	总结
层级关系	1 App = N Jobs = N×M Stages = N×M×P Tasks
Job 触发	每个 Action 算子调用 sc.runJob() 创建新 Job
Stage 切分	遇到 ShuffleDependency 即切分
Task 生成	Stage 最后一个 RDD 的 Partition 数量 = Task 数

金句：Transformation 是"画图纸"（构建 DAG），Action 是"按下启动键"（触发 Job）。一个 Application 可以画无数张图纸，但只有按下的启动键才算数。

---

作者：starzy | AI Data Engineer / 大数据技术实践者
博客：blog.starzy.cn | GitHub：starzy1990.github.io
专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践