2.1.2 Stage划分与任务生成

技术架构定位

Stage划分与任务生成是Spark执行引擎的核心环节，它将抽象的RDD转换图谱转化为实际的分布式计算任务。在Spark的计算流水线中，这一环节犹如建筑师将蓝图转化为具体施工方案的过程，既需要全局视野，又要考虑细节实现。

无法加载本地 PlantUML 图表

期望路径: /diagrams/22a6e0c4279faf36dcbffb34de67feb9_77373271.svg

Stage划分与任务生成处于Spark执行流程的关键位置，它连接了上层的RDD抽象操作与底层的具体任务执行。当用户代码触发Action操作时，Spark需要将逻辑执行计划（由RDD依赖关系组成的有向无环图）转换为物理执行计划（具体的计算任务序列）。这一转换过程由DAGScheduler负责，它首先将RDD图谱分解为多个Stage，然后为每个Stage生成对应的任务，最后将这些任务提交给TaskScheduler进行资源分配和执行调度。

这一环节的精妙之处在于，它将复杂的分布式计算问题分解为可管理的子问题，实现了"分而治之"的经典策略。通过合理的Stage划分，Spark能够最大化并行处理能力，同时通过任务的精确生成和调度，保证了计算结果的正确性和可靠性。本文将深入探讨这一关键环节的内部机制，揭示Spark如何将抽象的计算模型转化为高效的分布式执行方案。

DAG构建源码

在Spark执行过程的开端，DAG（有向无环图）的构建仿佛是一位建筑师在施工前精心绘制蓝图的过程。这个过程将用户的抽象转换操作序列转化为计算依赖关系图，为后续的Stage划分奠定基础。

无法加载本地 PlantUML 图表

期望路径: /diagrams/22a6e0c4279faf36dcbffb34de67feb9_64771672.svg

从RDD链到逻辑执行图

DAG构建的起点是用户触发Action操作时的终点RDD。与数学证明中从结论逆推的思路类似，Spark从需要计算的目标RDD出发，通过回溯其依赖关系，构建完整的逻辑执行图。这种方法确保了执行图中仅包含计算所必需的RDD，避免了不必要的计算和资源浪费。

DAG构建的核心代码位于DAGScheduler.handleJobSubmitted方法中。该方法接收最终RDD、计算函数和分区信息等参数，然后开始构建执行图。这一过程不仅要考虑RDD间的依赖关系，还需识别关键的"shuffle边界"，这些边界将标记不同Stage之间的分界线。

private[scheduler] def handleJobSubmitted(jobId: Int, finalRDD: RDD[_], func: (TaskContext, Iterator[_]) => _, ...) {
  var finalStage: ResultStage = null
  try {
    // 创建ResultStage（最终结果阶段）
    finalStage = newResultStage(finalRDD, func, partitions, jobId, callSite)
  } catch {
    case e: Exception =>
      logWarning("Creating new stage failed due to exception", e)
      listener.jobFailed(e)
      return
  }
  
  // 提交最终阶段，这将递归地提交所有依赖阶段
  submitStage(finalStage)
}

创建ResultStage过程中，Spark会回溯检查RDD依赖链，遇到宽依赖（ShuffleDependency）时会创建对应的ShuffleMapStage。这种回溯方式，犹如组装一个复杂的乐高积木，先确定最终形态，然后识别出需要哪些基础构件，以及它们的组装顺序。

值得注意的是，DAG构建过程是惰性的。当用户定义一系列RDD转换操作时，Spark并不会立即构建执行图，而是等到Action操作触发实际计算时才开始。这种设计与RDD的惰性计算特性一脉相承，允许Spark在掌握完整计算意图后进行全局优化。

依赖管理与Stage边界识别

在构建DAG的过程中，依赖关系的管理和Stage边界的识别是关键环节。Spark通过分析RDD的依赖类型（窄依赖或宽依赖），确定Stage的边界和执行顺序。这就像城市规划师划分功能区块，既要确保各区块内部功能协调，又要考虑区块间的交通和资源流动。

具体实现上，Spark使用getOrCreateParentStages方法识别当前RDD的所有父Stage。该方法会递归地分析RDD的依赖关系，遇到ShuffleDependency时创建新的Stage：

private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
  getShuffleDependencies(rdd).map { shuffleDep =>
    getOrCreateShuffleMapStage(shuffleDep, firstJobId)
  }.toList
}

private def getShuffleDependencies(rdd: RDD[_]): HashSet[ShuffleDependency[_, _, _]] = {
  val parents = new HashSet[ShuffleDependency[_, _, _]]
  val visited = new HashSet[RDD[_]]
  val waitingForVisit = new ArrayStack[RDD[_]]
  waitingForVisit.push(rdd)
  
  // 深度优先搜索遍历RDD依赖图
  while (waitingForVisit.nonEmpty) {
    val toVisit = waitingForVisit.pop()
    if (!visited(toVisit)) {
      visited += toVisit
      toVisit.dependencies.foreach {
        case shuffleDep: ShuffleDependency[_, _, _] =>
          parents += shuffleDep
        case dependency =>
          waitingForVisit.push(dependency.rdd)
      }
    }
  }
  parents
}

这个递归分析过程使用深度优先搜索（DFS）策略遍历RDD依赖图，标记所有ShuffleDependency作为Stage边界。DFS算法在这里的应用展现了Spark对计算机科学经典算法的灵活运用，使得复杂图结构的处理变得有序和高效。

值得一提的是，Spark会尝试重用之前已创建的Stage，这种优化避免了重复计算，特别是在复杂依赖关系或多次共享同一RDD的场景中尤为重要。getOrCreateShuffleMapStage方法中包含了查找现有Stage的逻辑，这种复用机制类似于动态规划中的记忆化搜索，避免了对相同子问题的重复求解：

private def getOrCreateShuffleMapStage(
    shuffleDep: ShuffleDependency[_, _, _],
    firstJobId: Int): ShuffleMapStage = {
  shuffleIdToMapStage.get(shuffleDep.shuffleId) match {
    case Some(stage) => stage
    case None =>
      // 创建新的ShuffleMapStage
      val rdd = shuffleDep.rdd
      val (parentStages, id) = getParentStagesAndId(rdd, firstJobId)
      val stage = new ShuffleMapStage(id, rdd, shuffleDep.partitioner.numPartitions, 
                                    shuffleDep, parentStages, taskIdToLocations, properties)
      // ...
      shuffleIdToMapStage(shuffleDep.shuffleId) = stage
      stage
  }
}

通过这种方式，Spark成功地将复杂的RDD依赖关系转化为有组织的Stage集合，每个Stage内部的计算可以流水线执行，而Stage之间则通过Shuffle操作进行数据交换。这种划分策略平衡了并行度和数据依赖，为高效执行奠定了基础。

Stage划分算法

Stage划分算法是Spark调度系统的精髓，它如同将复杂建筑的整体施工方案分解为多个独立可执行的施工阶段。通过巧妙地识别和处理依赖关系，Spark能够最大化并行执行并减少不必要的数据交换。

无法加载本地 PlantUML 图表

期望路径: /diagrams/22a6e0c4279faf36dcbffb34de67feb9_41348368.svg

基于宽依赖的反向遍历算法

Spark的Stage划分基于一个核心原则：以宽依赖（ShuffleDependency）为边界划分Stage，确保每个Stage内部只包含窄依赖，可以流水线执行。这种划分方法颇似河流的分水岭，自然地将计算流划分为不同的区域。

具体实现上，Spark采用基于宽依赖的反向遍历算法。从最终RDD开始，沿着依赖关系向上回溯。当遇到宽依赖时，会为该依赖创建一个新的ShuffleMapStage，然后继续回溯该依赖的父RDD。这一过程递归进行，直到处理完所有RDD依赖。

private def getParentStagesAndId(rdd: RDD[_], firstJobId: Int): (List[Stage], Int) = {
  val parentStages = getOrCreateParentStages(rdd, firstJobId)
  val id = nextStageId.getAndIncrement()
  (parentStages, id)
}

private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
  getShuffleDependencies(rdd).map { shuffleDep =>
    getOrCreateShuffleMapStage(shuffleDep, firstJobId)
  }.toList
}

Stage划分算法可以类比于项目管理中的关键路径分析。每个Stage代表一个必须完成的工作包，Stage之间的宽依赖表明了前后顺序约束。通过这种方式，Spark能够明确计算的进度和依赖条件，在保证结果正确性的前提下，最大化并行处理能力。

值得注意的是，Spark遵循"懒创建"原则，只有当一个Stage的所有父Stage都已完成时，才会提交该Stage执行。这种策略确保了资源不会浪费在无法立即执行的任务上，同时也减少了因依赖未满足导致的任务失败。

Stage类型与执行特点

Spark中存在两种类型的Stage：ShuffleMapStage和ResultStage，它们各自承担不同的责任并具有独特的执行特点。

ShuffleMapStage执行产生Shuffle数据的计算，其输出将被后续Stage消费。这类似于流水线生产中的中间环节，既需要处理上游输入，又需要为下游提供规格化的输出。每个ShuffleMapStage的边界由ShuffleDependency确定，其结果通常写入Shuffle系统以备后续使用。

ResultStage则是执行流的终点，直接计算Action操作的结果。它可能会产生返回给Driver的数据，或者执行向外部存储系统的写入操作。ResultStage犹如项目的最终交付环节，标志着特定计算流程的完成。

这两种Stage在实现上有许多共同点，都需要处理任务生成、失败重试和结果收集等问题，但也存在诸多差异：

class ShuffleMapStage(
    id: Int,
    rdd: RDD[_],
    numTasks: Int,
    shuffleDep: ShuffleDependency[_, _, _],
    parents: List[Stage],
    jobId: Int,
    callSite: CallSite) extends Stage(id, rdd, numTasks, parents, jobId, callSite) {
  
  // ShuffleMapStage特有的属性和方法
  val shuffleId: Int = shuffleDep.shuffleId
  var numAvailableOutputs: Int = 0
  // ...
}

class ResultStage(
    id: Int,
    rdd: RDD[_],
    func: (TaskContext, Iterator[_]) => _,
    numTasks: Int,
    parents: List[Stage],
    jobId: Int,
    callSite: CallSite) extends Stage(id, rdd, numTasks, parents, jobId, callSite) {
  
  // ResultStage特有的属性和方法
  val func: (TaskContext, Iterator[_]) => _
  // ...
}

Stage的执行顺序由它们的依赖关系决定，形成一个有向无环图。每个Stage只有在其所有父Stage都已成功完成时才能开始执行，这确保了数据依赖的正确满足。在执行过程中，Spark会追踪每个Stage的状态，包括等待执行、正在执行、执行成功或执行失败等。当某个Stage执行失败时，Spark会尝试重新执行该Stage，如果问题持续存在，则可能需要重新执行其父Stage以重建丢失的Shuffle数据。

通过这种分阶段执行的模式，Spark将复杂的计算过程分解为可管理的执行单元，既保证了执行的正确性和可靠性，又提供了灵活的调度和优化机会。

任务生成逻辑

任务是Spark执行的最小单位，它将抽象的计算描述转化为具体的执行指令。任务的生成过程犹如军队将战略计划转化为具体的战术行动，需要考虑地形（数据分布）、资源（计算能力）和目标（计算需求）等多种因素。

无法加载本地 PlantUML 图表

期望路径: /diagrams/22a6e0c4279faf36dcbffb34de67feb9_72882666.svg

ShuffleMapTask与ResultTask创建

Spark中的任务分为两种类型：ShuffleMapTask和ResultTask，分别对应于ShuffleMapStage和ResultStage。这种对应关系保证了任务的执行特性与Stage的需求相匹配。

任务创建的核心代码位于DAGScheduler的submitMissingTasks方法中。该方法首先确定Stage中哪些分区需要计算（可能是全部分区，也可能是之前执行失败的分区），然后为每个分区创建对应的任务实例：

private def submitMissingTasks(stage: Stage, jobId: Int) {
  // 获取需要计算的分区
  val partitionsToCompute: Seq[Int] = stage.findMissingPartitions()
  
  // 获取任务属性
  val properties = jobIdToActiveJob(jobId).properties
  
  // 根据Stage类型创建不同类型的任务
  val tasks: Seq[Task[_]] = stage match {
    case s: ShuffleMapStage =>
      partitionsToCompute.map { id =>
        val locs = getPreferredLocs(stage.rdd, id)
        val part = stage.rdd.partitions(id)
        new ShuffleMapTask(stage.id, stage.latestInfo.attemptId,
          taskBinary, part, locs, properties, serializedTaskMetrics, Option(jobId),
          Option(sc.applicationId), sc.applicationAttemptId, stage.rdd.isBarrier())
      }
      
    case s: ResultStage =>
      partitionsToCompute.map { id =>
        val p = stage.rdd.partitions(id)
        val locs = getPreferredLocs(stage.rdd, id)
        new ResultTask(stage.id, stage.latestInfo.attemptId,
          taskBinary, p, locs, id, properties, serializedTaskMetrics,
          Option(jobId), Option(sc.applicationId), sc.applicationAttemptId,
          stage.rdd.isBarrier())
      }
  }
  
  // 将任务集提交给TaskScheduler
  taskScheduler.submitTasks(new TaskSet(
    tasks.toArray, stage.id, stage.latestInfo.attemptId, jobId, properties))
}

ShuffleMapTask的主要职责是执行RDD转换操作并将结果写入Shuffle系统，其输出是一组文件或内存块，将被记录在MapStatus中返回给Driver。这些输出形成了后续Stage的输入数据。ShuffleMapTask就像工厂生产线上的中间工序，负责将原材料转化为半成品，并确保这些半成品被正确标记和存储，以便下一工序使用。

ResultTask则直接计算最终结果，其输出可能返回给Driver或写入外部系统。ResultTask相当于生产线的最后一道工序，直接生产出最终产品。与ShuffleMapTask不同，ResultTask的结果通常立即可见，无需等待其他任务的配合。

每个任务都包含必要的执行信息，如Stage ID、分区对象、任务二进制代码和位置偏好等。这些信息使得任务能够在分布式环境中独立执行，同时也提供了调度优化的机会。例如，位置偏好信息可以帮助TaskScheduler尽可能地将任务分配到数据所在的节点，减少网络传输开销。

任务封装与序列化

任务的封装和序列化是将计算逻辑从Driver节点传输到Executor节点的关键步骤。这一过程就像将军将作战计划分发给前线指挥官的过程，需要确保信息完整、清晰且安全。

任务封装的核心是taskBinary，它包含了执行任务所需的所有信息，包括RDD的依赖链、函数闭包和必要的配置。这个二进制包是通过序列化Java对象生成的，包含了计算逻辑与上下文：

// 准备任务的二进制表示
val taskBinary: Broadcast[Array[Byte]] = {
  // 函数闭包序列化
  val taskBinaryBytes: Array[Byte] = 
    if (stage.isInstanceOf[ShuffleMapStage]) {
      closureSerializer.serialize((stage.rdd, stage.shuffleDep.get): AnyRef).array()
    } else {
      closureSerializer.serialize((stage.rdd, stage.asInstanceOf[ResultStage].func): AnyRef).array()
    }
  
  // 广播序列化后的任务
  sc.broadcast(taskBinaryBytes)
}

序列化是分布式系统的常见挑战，Spark通过多种机制应对这一挑战：

1. 使用高效的序列化框架（如Kryo）减少序列化数据大小
2. 通过广播变量避免重复传输大型只读数据
3. 确保所有闭包中捕获的变量都是可序列化的

任务序列化失败是Spark应用中常见的错误之一，通常表现为"Task not serializable"异常。这种错误可能由各种原因引起，如闭包中捕获了不可序列化的对象（如数据库连接、文件句柄等）。理解任务封装和序列化机制有助于诊断和解决这类问题。

值得注意的是，任务的序列化不仅关乎功能正确性，还影响执行效率。过大的序列化任务会增加网络传输开销和反序列化时间，从而降低整体性能。因此，设计合理的转换函数和避免捕获不必要的大对象是Spark应用优化的重要方面。

任务提交流程

任务从创建到实际执行需要经过多个环节，这一流程就像物流系统将货物从仓库配送到目的地的过程，涉及多层级的协调和处理。了解任务提交流程有助于理解Spark执行模型和定位潜在问题。

无法加载本地 PlantUML 图表

期望路径: /diagrams/22a6e0c4279faf36dcbffb34de67feb9_20448126.svg

从DAGScheduler到TaskScheduler的交互

任务提交的第一步是从DAGScheduler到TaskScheduler的交互。这一步将Stage级别的任务集（TaskSet）提交给TaskScheduler，由后者负责细粒度的任务调度。这种分层设计体现了"关注点分离"的软件工程原则，使得系统各部分能够专注于自己的职责。

DAGScheduler通过submitTasks方法将TaskSet提交给TaskScheduler：

taskScheduler.submitTasks(new TaskSet(
  tasks.toArray, stage.id, stage.latestInfo.attemptId, jobId, properties))

TaskSet包含了同一Stage中的多个任务，这些任务可以并行执行，因为它们处理的是不同的数据分区，没有直接依赖关系。接收到TaskSet后，TaskScheduler会创建一个TaskSetManager来管理该TaskSet的生命周期：

// 在TaskScheduler.submitTasks中
val manager = createTaskSetManager(taskSet, maxTaskFailures)
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)

TaskSetManager负责任务的调度、重试和结果跟踪等工作。它实现了细粒度的任务管理，如根据本地性偏好分配任务、处理任务失败和管理推测执行等。这一设计使得TaskScheduler能够灵活地适应不同的调度策略和执行环境，而不影响上层的DAG处理逻辑。

调度策略实现

任务的实际调度由TaskScheduler的scheduleTask方法和TaskSetManager的resourceOffer方法共同完成。这一过程考虑了多种因素，如数据本地性、任务优先级和公平共享等，旨在优化整体执行效率和资源利用率。

调度过程的核心是资源分配循环。当有可用的执行资源（如Executor上的空闲核心）时，TaskScheduler会按照特定顺序轮询各个TaskSetManager，请求它们提供适合该资源的任务：

private def resourceOfferSingleTaskSet(
    taskSet: TaskSetManager,
    maxLocality: TaskLocality,
    tasks: IndexedSeq[Int],
    availableCpus: Array[Int],
    availableResources: Array[Map[String, Buffer[String]]],
    availableSlots: Array[Buffer[TaskLocation]]): Boolean = {
  
  var launchedTask = false
  for (i <- 0 until availableCpus.length if !launchedTask) {
    val execId = s"executor_${i}"
    val host = s"host_${i}"
    val taskSetId = taskSet.taskSet.id
    
    taskSet.resourceOffer(execId, host, maxLocality, availableResources(i)) match {
      case Some(task) =>
        taskIdToTaskSetManager(task.taskId) = taskSet
        taskIdToExecutorId(task.taskId) = execId
        executorIdToTaskCount(execId) += 1
        availableCpus(i) -= 1
        assert(availableCpus(i) >= 0)
        launchedTask = true
        // ...
        
      case None => // 没有合适的任务
    }
  }
  launchedTask
}

TaskSetManager的resourceOffer方法实现了数据本地性优化。Spark定义了多个本地性级别，如PROCESS_LOCAL（同一JVM内）、NODE_LOCAL（同一节点内）、RACK_LOCAL（同一机架内）和ANY（任意位置）。调度器会尝试按照从最佳到最差的顺序分配任务，但会设置等待超时以避免资源闲置：

def resourceOffer(
    execId: String,
    host: String,
    maxLocality: TaskLocality.TaskLocality,
    taskResourceAssignments: Map[String, Seq[String]] = Map.empty)
  : Option[TaskDescription] = {
  
  if (freeTasks.isEmpty) {
    return None
  }
  
  for (index <- dequeueTask(execId, host, maxLocality)) {
    val task = tasks(index)
    val taskId = sched.nextTaskId()
    // 创建任务描述
    val taskDesc = new TaskDescription(taskId, attemptNum = currentAttemptId,
      execId, taskName, index, task.serializedTask, task.addedFiles,
      task.addedJars, task.properties, task.resources)
    return Some(taskDesc)
  }
  None
}

Spark的调度策略还包括对失败任务的重试、推测执行（执行多个相同任务副本以减少长尾影响）和资源动态分配等高级功能。这些策略共同构成了一个强大而灵活的执行框架，能够适应各种工作负载和执行环境。

通过这种多层次的调度系统，Spark能够将抽象的计算描述转化为具体的执行命令，并在分布式环境中有效地协调资源和任务。任务提交流程的每一步都体现了分布式系统设计的智慧和折衷，是Spark执行模型的核心组成部分。

调度策略优化

Spark的调度策略不仅要保证计算的正确性，还追求执行效率的最大化。通过灵活的调度策略和精心的优化，Spark能够自适应地处理各种工作负载，就像一位优秀的项目经理能够根据不同任务的特点和团队成员的能力，灵活地分配和调整工作计划。

无法加载本地 PlantUML 图表

期望路径: /diagrams/22a6e0c4279faf36dcbffb34de67feb9_89614943.svg

本地性感知与资源匹配算法

数据本地性是分布式计算中的关键优化目标。在大数据环境中，移动计算比移动数据更经济，因此Spark采用本地性感知调度算法，尽可能地将任务分配到数据所在的位置。这种策略就像在餐厅分配服务员时，优先让每个服务员负责靠近自己区域的桌台，减少不必要的走动和交叉。

Spark定义了五个本地性级别，从最优到最差依次是：

• PROCESS_LOCAL：数据在同一JVM内，无需数据传输
• NODE_LOCAL：数据在同一节点但不同JVM内，需要进程间通信
• RACK_LOCAL：数据在同一机架内的不同节点，需要通过网络传输但带宽较高
• ANY：任意位置，可能需要跨机架或跨数据中心的网络传输

TaskSetManager的调度逻辑会尝试按照从最优到最差的顺序分配任务。当找不到满足最优本地性级别的任务时，它不会立即降级到次优级别，而是会等待一段时间，以期有更好的资源出现：

private def dequeueTask(execId: String, host: String, locality: TaskLocality.Value)
  : Option[Int] = {
  
  def findTask(locality: TaskLocality.Value): Option[Int] = {
    val localityIndex = TaskLocality.isAllowed(locality, maxLocality)
    if (localityIndex >= 0) {
      // 检查是否应该跳过当前本地性级别
      val curTime = clock.getTimeMillis()
      if (TaskLocality.isLocal(maxLocality) && !isLocal(locality) &&
          curTime - lastLaunchTime < localityWaits(localityIndex)) {
        // 还在等待更好的本地性，跳过当前级别
        defer(host, locality)
        None
      } else {
        // 已超过等待时间或找到了满足条件的任务
        findTaskFromList(execId, host, pendingTasksForExecutor, locality) orElse
        findTaskFromList(execId, host, pendingTasksForHost, locality) orElse
        findTaskFromList(execId, host, pendingTasksWithNoPrefs, locality) orElse
        findTaskFromList(execId, host, pendingTasksForRack, locality) orElse
        findTaskFromList(execId, host, allPendingTasks, locality)
      }
    } else {
      None
    }
  }
  
  // 按本地性级别尝试分配任务
  findTask(locality) orElse
  findTask(TaskLocality.NODE_LOCAL) orElse
  findTask(TaskLocality.RACK_LOCAL) orElse
  findTask(TaskLocality.ANY)
}

本地性等待时间是可配置的，用户可以根据自己的集群特性和应用需求调整相关参数，如spark.locality.wait、spark.locality.wait.process等。这些参数决定了在降级到较差本地性级别前等待的时间，较长的等待时间可能提高数据本地性，但也可能导致资源闲置；较短的等待时间则可能提高资源利用率，但增加网络传输开销。

除了本地性感知外，Spark还实现了资源匹配算法，考虑任务对CPU、内存和其他资源的需求。在Spark 3.0之后，引入了更细粒度的资源分配机制，支持任务级别的资源请求和分配，这使得异构环境下的调度更为高效。

推测执行与故障恢复

在大规模分布式环境中，节点性能差异和临时故障是常态。为了减少这些因素对整体执行时间的影响，Spark实现了推测执行和故障恢复机制。这些机制就像赛车比赛中的备用轮胎和紧急维修，确保即使在条件不理想的情况下也能顺利完成比赛。

推测执行（Speculative Execution）是一种应对长尾任务的策略。当一个Stage中的大部分任务已经完成，但少数任务仍在执行且进度明显落后时，Spark会为这些慢任务启动备份任务。无论哪个任务版本先完成，其结果都会被采用，而另一版本会被取消。这种机制有效减少了因个别慢节点导致的执行延迟：

def checkSpeculatableTasks(minTimeToSpeculation: Int): Boolean = {
  // 计算平均执行时间和标准差
  val runningTaskTimes = runningTasks.map { case (tid, info) =>
    info.timeRunning(clock.getTimeMillis())
  }
  if (runningTaskTimes.nonEmpty) {
    val mean = runningTaskTimes.sum / runningTaskTimes.size
    val threshold = max(SPECULATION_MULTIPLIER * mean, minTimeToSpeculation)
    
    // 找出执行时间超过阈值的任务
    val speculatableTasks = runningTasks.filter { case (tid, info) =>
      info.timeRunning(clock.getTimeMillis()) > threshold &&
      !successful(info.index) &&
      !speculationScheduled(info.index)
    }
    
    // 启动推测任务
    speculatableTasks.keys.foreach { tid =>
      addPendingTask(tid)
      speculationScheduled(tid) = true
    }
    speculatableTasks.nonEmpty
  } else {
    false
  }
}

推测执行可通过spark.speculation参数启用，并可使用spark.speculation.multiplier和spark.speculation.quantile等参数调整触发条件。在资源充足且任务执行时间差异大的场景中，推测执行能显著提升性能；但在资源紧张或任务重复执行代价高的场景中，可能需要禁用或谨慎配置该特性。

故障恢复是Spark容错机制的核心。当任务执行失败时，Spark会根据失败原因和重试次数采取不同的恢复策略：

• 任务级别重试：如果单个任务失败且未超过最大重试次数，仅重试该任务
• Stage级别重试：如果特定任务反复失败或Stage中的多个任务失败，重新执行整个Stage
• Job级别失败：如果重试无法解决问题或超过最大重试次数，整个Job失败

这种多级恢复策略平衡了容错性和效率，既能处理临时故障，又能避免在永久性问题上浪费资源。结合RDD的血缘关系，Spark能够精确地重建丢失的数据分区，最小化故障影响范围。

通过本地性感知、推测执行和故障恢复等高级调度策略，Spark能够在复杂多变的分布式环境中高效执行数据处理任务，实现了性能与可靠性的双赢。这些机制体现了分布式系统设计的艺术，是Spark成功的关键因素之一。

技术关联

Stage划分与任务生成是Spark执行引擎的核心环节，它与整个分布式计算生态系统中的多个关键技术紧密关联。理解这些技术联系有助于全面掌握Spark的执行机制和优化方法。

无法加载本地 PlantUML 图表

期望路径: /diagrams/22a6e0c4279faf36dcbffb34de67feb9_87031386.svg

与RDD内部实现的紧密联系

Stage划分与任务生成与RDD内部实现有着不可分割的联系。RDD的依赖关系直接决定了Stage的划分边界，而RDD的分区结构则映射为具体的任务划分。这种联系体现了Spark设计的一致性与整体性，各组件间相互配合，构成完整的执行流程。

RDD的窄依赖和宽依赖概念是Stage划分的理论基础。窄依赖允许父RDD和子RDD的分区在同一Stage内流水线执行，而宽依赖则要求在Stage间进行Shuffle操作。这种依赖关系的分类直接影响了Spark如何组织和优化计算。

RDD的计算函数（compute方法）在任务执行时被调用，用于生成实际的数据。任务的核心逻辑就是调用相应分区的计算函数，并处理其输出结果。因此，RDD的计算模型与任务执行模型紧密绑定，共同决定了Spark的执行行为。

RDD的首选位置（preferredLocations方法）为任务调度提供了本地性信息。TaskScheduler利用这些信息尝试将任务分配到数据所在位置，减少数据传输开销。这种数据本地性优化是分布式系统中的常见策略，也是Spark性能优势的重要来源。

与Shuffle系统的交互

Stage划分与Shuffle系统的交互是Spark执行流程中最重要的环节之一。每个Stage的边界通常是由Shuffle操作定义的，而Shuffle过程则是不同Stage之间交换数据的桥梁。这种交互关系决定了数据如何在集群中流动和重组，直接影响执行效率和资源利用率。

在DAG构建阶段，Spark识别ShuffleDependency并以此为边界划分Stage。每个ShuffleDependency对应一个shuffleId，用于唯一标识特定的Shuffle操作。这些ID在后续的任务执行中用于定位Shuffle数据。

ShuffleMapTask的输出写入Shuffle系统，形成下一Stage的输入数据。这个过程涉及数据的分区、序列化、写磁盘和网络传输等多个环节，是Spark性能优化的关键领域。

Shuffle读取的效率直接影响Stage的启动延迟。如果Shuffle数据尚未完全可用，下游Stage的任务可能需要等待，这就形成了Stage间的执行依赖。Spark的调度系统会追踪这些依赖关系，确保各Stage按正确顺序执行。

与调度和资源管理系统的配合

Stage划分与任务生成需要与调度和资源管理系统紧密配合，共同实现高效的分布式执行。这种配合体现在任务的创建、提交、调度和监控等多个环节，形成完整的执行闭环。

TaskScheduler负责将DAGScheduler生成的任务分配给具体的执行节点。它考虑多种因素，如数据本地性、资源需求和任务优先级等，尝试找到最优的任务放置方案。

不同的集群管理器（如YARN、Mesos和Kubernetes）提供各自的资源抽象和分配机制。Spark通过SchedulerBackend适配这些系统，将任务调度转化为特定环境下的资源请求和任务部署。

动态资源分配是Spark的高级特性，它允许应用根据负载变化自动申请和释放资源。这一特性需要任务生成系统与资源管理器紧密协作，共同实现资源的弹性伸缩。

对高级功能和优化的启发

Stage划分与任务生成不仅是Spark基本功能的核心，也是多项高级功能和优化策略的基础。这些特性和优化扩展了Spark的应用范围并提升了性能，体现了系统的可扩展性和适应性。

SQL优化器（Catalyst）将SQL查询转换为最优的物理执行计划。这一过程需要考虑Stage划分和任务执行的特性，如利用窄依赖的流水线执行和优化Shuffle操作的代价等。

自适应查询执行（AQE）允许Spark在运行时根据统计信息动态调整执行计划。这一特性需要能够重新划分Stage和生成新的任务，对执行引擎的灵活性提出了更高要求。

数据倾斜处理是Spark性能优化的重要方面。通过识别倾斜的Shuffle操作并应用特殊的处理策略（如拆分倾斜键或局部聚合等），可以显著提升执行效率。这些策略通常涉及修改Stage划分和任务生成逻辑。

从更广泛的角度看，Stage划分与任务生成是分布式计算的核心问题之一，涉及如何将抽象的计算描述转化为具体的执行指令，以及如何在分布式环境中协调这些指令的执行。Spark在这一领域的探索和实践，不仅推动了自身的发展，也为其他分布式系统提供了宝贵经验和参考。

通过理解Stage划分与任务生成的技术关联，我们可以更全面地掌握Spark的执行机制，并在设计和优化Spark应用时做出更明智的决策。无论是优化现有功能还是扩展新特性，这些理解都是不可或缺的基础。

参考资料

[1] Matei Zaharia et al. Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing. NSDI 2012.

[2] Apache Spark 官方文档. https://spark.apache.org/docs/latest/job-scheduling.html

[3] Holden Karau, Andy Konwinski, Patrick Wendell, Matei Zaharia. Learning Spark: Lightning-Fast Big Data Analysis. O’Reilly Media, 2015.

[4] Kay Ousterhout et al. Making Sense of Performance in Data Analytics Frameworks. NSDI 2015.

[5] Aaron Davidson, Andrew Or. Optimizing Shuffle Performance in Spark. Technical Report, University of California, Berkeley, 2013.

被引用于

[1] Spark-Shuffle实现机制

[2] Spark-容错机制实现

[3] Spark-Catalyst优化器实现

[4] Spark-数据倾斜处理实践