夜雨聆风学习资料网

ARTICLE · 997892

【极速学习spark源码】Task 在 Executor 上的完整生命周期

发布时间：2026-09-15 22:32:25 最近访问：2026-09-15 22:32:24
【极速学习spark源码】Task 在 Executor 上的完整生命周期

基于 Spark 4.2，分支 branch-4.2

阅读时长约 12 分钟 · 进阶

背景

前面我们已经看到 Driver 端如何创建 TaskDescription，并通过 SchedulerBackend 发给 Executor。

那 Executor 收到 Task 后发生了什么？

这篇文章沿着 Executor 端代码走一遍：

LaunchTask
  → Executor.launchTask
    → TaskRunner.run
      → Task.run
        → ResultTask.runTask / ShuffleMapTask.runTask
          → 状态回传 Driver

一、入口：LaunchTask 消息

Driver 发给 Executor 的 RPC 消息定义在 CoarseGrainedClusterMessage.scala:45：

case class LaunchTask(data: SerializableBuffer) extends CoarseGrainedClusterMessage

Executor 端收到消息的位置在 CoarseGrainedExecutorBackend.scala:181-187：

case LaunchTask(data) =>  val taskDesc = TaskDescription.decode(data.value)  executor.launchTask(this, taskDesc)

从这里开始，任务进入真正的 Executor 执行流程。

二、Executor.launchTask：创建 TaskRunner

Executor.scala:249-257 是 Executor 类定义和说明。

入口方法在 Executor.scala:551-563：

def launchTask(context: ExecutorBackend, taskDescription: TaskDescription): Unit = {  val tr = createTaskRunner(context, taskDescription)  runningTasks.put(taskId, tr)  val killMark = killMarks.get(taskId)  if (killMark != null) {    tr.kill(killMark._1, killMark._2)    killMarks.remove(taskId)  }  threadPool.execute(tr)}

这一步只做三件事：

  1. 创建 TaskRunner

  2. 放入 runningTasks

  3. 丢进线程池执行

三、TaskRunner：Executor 端的任务线程

Executor.scala:687-691：

class TaskRunner(    execBackend: ExecutorBackend,    val taskDescription: TaskDescription,    private val plugins: Option[PluginContainer]) extends Runnable

真正入口在 Executor.scala:806：

override def run(): Unit = {  // 初始化上下文、反序列化 task、调用 task.run、处理结果或异常}

TaskRunner.run 做的事情很多，可以分成 6 段：

  1. 初始化线程名、classloader、serializer

  2. 回传 TaskState.RUNNING（Executor.scala:830）

  3. 反序列化 taskDescription.serializedTask（Executor.scala:841-850）

  4. 调用 task.run(...)（Executor.scala:887-895）

  5. 序列化结果并决定直接/间接回传（Executor.scala:940-1023）

  6. 回传 TaskState.FINISHED（Executor.scala:1029）

四、Task.run：创建 TaskContext

Task.scala:61-74：

private[spark] abstract class Task[T](...)

Task.run 入口在 Task.scala:87-93：

final def run(    taskAttemptId: Long,    attemptNumber: Int,    metricsSystem: MetricsSystem,    cpus: Int,    resources: Map[String, ResourceInformation],    plugins: Option[PluginContainer]): T = {

它会创建 TaskContextImpl。Task.scala:101-117：

val taskContext = new TaskContextImpl(  stageId,  stageAttemptId,  partitionId,  taskAttemptId,  attemptNumber,  numPartitions,  taskMemoryManager,  localProperties,  metricsSystem,  metrics,  cpus,  resources)

如果是 barrier stage，会包装成 BarrierTaskContext（Task.scala:119-123）。

最后，Task.scala:146-148 调用：

context.runTaskWithListeners(this)

真正执行的抽象方法是 Task.scala:178：

def runTask(context: TaskContext): T

五、ResultTask：执行 action 的 func

ResultTask.scala:55-71 定义 ResultTask。

入口在 ResultTask.scala:78：

override def runTask(context: TaskContext): U = {  val ser = SparkEnv.get.closureSerializer.newInstance()  val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T]) => U)](    ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)  func(context, rdd.iterator(partition, context))}

这段特别重要：

  • Driver 端序列化的是 (RDD, func)

  • Executor 端反序列化出来

  • 调用 rdd.iterator(partition, context) 真正计算该分区

  • 把 iterator 交给 action 的 func

collect、count、take 这类最终返回结果给 Driver 的任务，都走 ResultTask。

六、ShuffleMapTask：写 shuffle 输出

ShuffleMapTask.scala:54-70 定义 ShuffleMapTask。

入口在 ShuffleMapTask.scala:82：

override def runTask(context: TaskContext): MapStatus = {  val ser = SparkEnv.get.closureSerializer.newInstance()  val rddAndDep = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])](    ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)  val rdd = rddAndDep._1  val dep = rddAndDep._2  dep.shuffleWriterProcessor.write(    rdd.iterator(partition, context), dep, mapId, partitionId, context)}

它不把用户结果回传给 Driver，而是：

  1. 计算当前 map partition

  2. 写 shuffle 文件

  3. 返回 MapStatus

MapStatus 告诉 Driver：这个 map 输出在哪里、每个 reduce 分区多大。

七、结果回传：DirectTaskResult / IndirectTaskResult

Task 执行完后，Executor 要把结果传回 Driver。

Executor.scala:940-944：先用 serializer 序列化任务返回值。

Executor.scala:988-992：构造 DirectTaskResult。

TaskResult.scala:33-44 定义了结果类型：

sealed trait TaskResult[T]case class DirectTaskResult[T](...) extends TaskResult[T]case class IndirectTaskResult[T](blockId: BlockId, size: Long) extends TaskResult[T]

如果结果较小，直接通过 RPC 回传；如果较大，写入 BlockManager，再回传一个 IndirectTaskResult（Executor.scala:1001-1023）。

成功回传在 Executor.scala:1029：

execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult)

八、Driver 端怎么接收结果

Executor backend 把状态发给 Driver：

  • ExecutorBackend.scala:27-28：statusUpdate(taskId, state, data) 接口

  • CoarseGrainedExecutorBackend.scala:267-279：把状态包装成 StatusUpdate 发给 Driver

  • CoarseGrainedSchedulerBackend.scala:168-179：Driver 收到后调用 scheduler.statusUpdate(...)

成功结果由 TaskResultGetter 处理：

  • TaskResultGetter.scala:65-78：反序列化 DirectTaskResult

  • TaskResultGetter.scala:78-105：遇到 IndirectTaskResult，从 BlockManager 拉取真实结果

  • TaskResultGetter.scala:111-122：交给 scheduler.handleSuccessfulTask

然后 TaskScheduler 通知 DAGScheduler，最终走到 DAGScheduler.handleTaskCompletion。

九、失败和 kill 路径

Task 不是总能成功。

kill

  • CoarseGrainedClusterMessage.scala:47-48：KillTask 消息

  • CoarseGrainedExecutorBackend.scala:190-194：收到 kill 后调用 executor.killTask(...)

  • Executor.scala:603-607：查找正在运行的 TaskRunner

  • Executor.scala:721-733：TaskRunner.kill

  • Task.scala:231-249：Task.kill

  • TaskContextImpl.scala:326-335：标记 interrupted

  • TaskContextImpl.scala:339-343：已中断时抛 TaskKilledException

普通失败

  • TaskContext.scala:199-207：runTaskWithListeners 先检查 kill，再执行 runTask

  • TaskContext.scala:208-222：异常时触发 failure/completion listeners

  • Executor.scala:1094-1112：普通异常转成 ExceptionFailure，回传 TaskState.FAILED

十、metrics 更新

Executor 端会记录大量指标：

  • Executor.scala:947-963：task deserialize time、run time、CPU time、GC time、result serialization、peak memory

  • Executor.scala:966-982：把 task metrics 累加到 executor metrics source

  • Executor.scala:1139-1184：shuffle read/write metrics

  • Executor.scala:1122-1129：TaskRunner finally 清理 runningTasks，并通知 metricsPoller task completion

所以 Spark UI 里看到的 task time、shuffle read/write、spill、GC 等，大多来自这里。

十一、实战：观察 Task 级指标

val rdd = sc.parallelize(1 to 1000000, 4)  .map(_ + 1)  .filter(_ % 2 == 0)println(rdd.count())

打开 Spark UI 的 Stage 页面：

http://localhost:4040/stages/

结果：

  • 每个 partition 对应一个 Task

  • 可以看到 Executor Run Time、Executor CPU Time、GC Time

  • 如果涉及 shuffle，还能看到 Shuffle Read / Write

  • 本地模式下 executor 可能显示为 driver 或 local executor

十二、总结

Executor 端 Task 生命周期：

Driver
  → LaunchTask(TaskDescription)
ExecutorBackend
  → Executor.launchTask
    → TaskRunner
      → statusUpdate(RUNNING)
      → deserialize Task
      → task.run
        → TaskContextImpl
        → ResultTask.runTask / ShuffleMapTask.runTask
      → serialize result
      → DirectTaskResult / IndirectTaskResult
      → statusUpdate(FINISHED / FAILED / KILLED)
Driver
  → TaskResultGetter
  → TaskScheduler.statusUpdate
  → DAGScheduler.handleTaskCompletion

三个认知：

  1. Executor 运行的是 TaskRunner，不是直接运行 Task。 TaskRunner 负责反序列化、状态回传、metrics、异常处理。

  2. ResultTask 和 ShuffleMapTask 的 runTask 完全不同。 一个返回用户结果，一个写 shuffle 并返回 MapStatus。

  3. Task 结果可能直接回传，也可能间接回传。 大结果会先写 BlockManager，再让 Driver 通过 IndirectTaskResult 拉取。

每天花费5分钟学习spark，让你技术之路走得更稳、更快。

喜欢的点个关注。

相关学习资料