> **摘要**：如果说 Job 是 Spark 的"任务单"，Stage 就是"施工阶段"，Task 就是每个工人的"具体活"。一个 Job 被 DAGScheduler 沿 Shuffle 边界切分为多个 Stage——前面的全是 ShuffleMapStage，最后一个必须是 ResultStage。每个 Stage 的 Partition 数决定了 Task 数量，ShuffleMapStage 产生 ShuffleMapTask（写 Shuffle 文件），ResultStage 产生 ResultTask（直接返回结果）。本文从 Stage 类型体系、DAG → Stage 切分源码、Task 生成与序列化、两种 Task 执行差异四个维度，配合 1 张原创深色架构图 + 完整源码分析，带你彻底看懂 Spark 最核心的执行引擎。
>
> **关键词**：Spark Stage, ShuffleMapStage, ResultStage, ShuffleMapTask, ResultTask, DAGScheduler, Task 序列化, MapOutputTracker

---

## 一、开篇：Stage 和 Task 是什么关系？

先说结论：

```
Job = 用户的一个 Action 操作
  ├── Stage 0: ShuffleMapStage → 2 个 ShuffleMapTask
  └── Stage 1: ResultStage     → 3 个 ResultTask
```

| 概念 | 定义 | 数量 |
|------|------|------|
| **Stage** | Shuffle 边界切分的计算阶段 | 每个 Job 可有多个 |
| **Task** | 处理一个 Partition 的最小计算单元 | 每个 Stage 可有多个 |
| **ShuffleMapStage** | 输出 Shuffle 中间文件的 Stage | Job 中除最后一个外的所有 |
| **ResultStage** | 输出最终结果的 Stage | 每个 Job 有且仅有一个 |

---

## 二、Stage 与 Task 全景图
![在这里插入图片描述](https://developer.qcloudimg.com/http-save/yehe-2482016/feeb7ed80c5aecb2e2cb99d25be82ba2.png)
## 三、Stage 切分：从 RDD DAG 到 Stage

### 3.1 核心源码

```scala
// 源码：DAGScheduler.scala - 创建 ResultStage
private def createResultStage(finalRDD: RDD[_], func: (TaskContext, Iterator[_]) => _,
    partitions: Array[Int], jobId: Int, callSite: CallSite): ResultStage = {
  // 从 finalRDD 回溯 → 遇到 ShuffleDep → 创建 ShuffleMapStage
  val parents = getOrCreateParentStages(finalRDD, jobId)
  val id = nextStageId.getAndIncrement()
  new ResultStage(id, finalRDD, func, partitions, parents, jobId, callSite)
}

// 递归获取父 Stage
private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
  rdd.dependencies.flatMap {
    case shufDep: ShuffleDependency[_, _, _] =>
      getOrCreateShuffleMapStage(shufDep, firstJobId) :: Nil
    case _ => Nil  // NarrowDep 不切分
  }.toList
}
```

### 3.2 Stage 提交顺序

```scala
// 递归提交：先父后子
private def submitStage(stage: Stage): Unit = {
  val missing = getMissingParentStages(stage).sortBy(_.id)
  if (missing.isEmpty) {
    submitMissingTasks(stage, jobId.get)  // 无缺失父 Stage → 执行
  } else {
    for (parent <- missing) submitStage(parent)  // 递归提交父 Stage
  }
}
```

---

## 四、Task 生成：从 Stage 到 TaskSet

```scala
// 源码：DAGScheduler.scala - submitMissingTasks()
private def submitMissingTasks(stage: Stage, jobId: Int): Unit = {
  // 计算需要计算的 Partition（跳过已完成的）
  val partitionsToCompute = stage.findMissingPartitions()

  // 为每个 Partition 创建一个 Task
  val tasks: Seq[Task[_]] = stage match {
    case stage: ShuffleMapStage =>
      partitionsToCompute.map { id =>
        new ShuffleMapTask(stage.id, stage.rdd, stage.shuffleDep, ...)
      }
    case stage: ResultStage =>
      partitionsToCompute.map { id =>
        new ResultTask(stage.id, stage.rdd, stage.func, id, ...)
      }
  }

  // 封装为 TaskSet，提交给 TaskScheduler
  taskScheduler.submitTasks(new TaskSet(tasks.toArray, stage.id, ...))
}
```

**Task 数量 = Stage 最后一个 RDD 的 Partition 数量。**

---

## 五、两种 Stage 与两种 Task 对比

### 5.1 ShuffleMapStage + ShuffleMapTask

```scala
// ShuffleMapTask.runTask() — 执行逻辑
override def runTask(context: TaskContext): MapStatus = {
  val writer = new ShuffleWriter(partition, shuffleDep)
  // ① 执行 RDD 算子链（map/flatMap/filter...）
  val iter = rdd.iterator(partition, context)
  // ② 将结果写入 Shuffle 文件
  writer.write(iter)
  // ③ 返回 MapStatus（文件位置 + 分区长度）
  writer.stop(success = true).get
}
```

### 5.2 ResultStage + ResultTask

```scala
// ResultTask.runTask() — 执行逻辑
override def runTask(context: TaskContext): U = {
  // ① 执行 RDD 算子链
  val iter = rdd.iterator(partition, context)
  // ② 将最终结果应用 func（如 collect 的收集逻辑）
  func(context, iter)
  // ③ 序列化结果 → StatusUpdate → Driver
}
```

### 5.3 对比表

| 维度 | ShuffleMapStage | ResultStage |
|------|----------------|-------------|
| Task 类型 | ShuffleMapTask | ResultTask |
| 输出 | Shuffle 中间文件 | 最终计算结果 |
| 返回类型 | MapStatus | U (泛型) |
| 一个 Job 中的数量 | 0~N | 1（唯一） |

---

## 六、Task 序列化

```bash
# 推荐 Kryo 序列化（比 Java 快 10 倍）
--conf spark.serializer=org.apache.spark.serializer.KryoSerializer
--conf spark.kryo.registrationRequired=true  # 强制注册
```

```scala
// 代码中注册 Kryo 类
val conf = new SparkConf()
  .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
  .registerKryoClasses(Array(classOf[MyDataClass], classOf[MyModel]))
```

**为什么需要序列化？** Driver 端的 Task 对象（包含 RDD 算子闭包）需要跨网络发送到 Executor，必须序列化为字节流。

---

## 七、总结

| 要点 | 总结 |
|------|------|
| **Stage 切分** | 遇到 ShuffleDependency 即切分，递归提交（先父后子） |
| **Task 生成** | 每个 Partition → 一个 Task，类型由 Stage 决定 |
| **两种 Stage** | ShuffleMapStage（写 Shuffle） + ResultStage（返回结果） |
| **序列化** | Task 闭包必须可序列化，推荐 Kryo |

> **金句**：Stage 是 Spark 的"流水线工位"，Task 是每个工位上的"工人"。Shuffle 就是工位之间的传送带——上一个工位写完，下一个工位才能开始。

---

<p style="margin-top:20px;padding-top:10px;border-top:1px solid #eee;font-size:14px;color:#999;text-align:center;line-height:2;">
作者：starzy  |  AI Data Engineer / 大数据技术实践者<br/>
博客：blog.starzy.cn  |  GitHub：starzy1990.github.io<br/>
专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践
</p>

摘要：如果说 Job 是 Spark 的"任务单"，Stage 就是"施工阶段"，Task 就是每个工人的"具体活"。一个 Job 被 DAGScheduler 沿 Shuffle 边界切分为多个 Stage——前面的全是 ShuffleMapStage，最后一个必须是 ResultStage。每个 Stage 的 Partition 数决定了 Task 数量，ShuffleMapStage 产生 ShuffleMapTask（写 Shuffle 文件），ResultStage 产生 ResultTask（直接返回结果）。本文从 Stage 类型体系、DAG → Stage 切分源码、Task 生成与序列化、两种 Task 执行差异四个维度，配合 1 张原创深色架构图 + 完整源码分析，带你彻底看懂 Spark 最核心的执行引擎。

Spark 核心之 Stage 和 Task 原理剖析

摘要：如果说 Job 是 Spark 的"任务单"，Stage 就是"施工阶段"，Task 就是每个工人的"具体活"。一个 Job 被 DAGScheduler 沿 Shuffle 边界切分为多个 Stage——前面的全是 ShuffleMapStage，最后一个必须是 ResultStage。每个 Stage 的 Partition 数决定了 Task 数量，ShuffleMapStage 产生

人工智能

Spark执行引擎的核心机制详解：Job经DAGScheduler沿Shuffle边界切分为多个Stage，除最后一个ResultStage外均为ShuffleMapStage。每个Partition对应一个Task，ShuffleMapTask负责写入Shuffle中间文件，ResultTask返回最终结果。文章从Stage类型体系、DAG切分源码、Task生成序列化到两种Task执行差异，配合架构图和完整源码分析，深度解析Stage与Task的协作逻辑及Task序列化最佳实践（推荐Kryo）。

数据架构

代码解释

大数据

LangGraph

GitHub

Spark 

Java

2026上云采购 | AI焕新·智启新局

bigdata-class

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

Spark执行引擎的核心机制详解：Job经DAGScheduler沿Shuffle边界切分为多个Stage，除最后一个ResultStage外均为ShuffleMapStage。每个Partition对应一个Task，ShuffleMapTask负责写入Shuffle中间文件，ResultTask返回最终结果。文章从S...

Spark 核心之 Stage 和 Task 原理剖析-腾讯云开发者社区-腾讯云

Spark 核心之 Stage 和 Task 原理剖析

Spark 核心之 Stage 和 Task 原理剖析

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐