在生产环境下， Spark 集群的部署方式一般为 YARN-Cluster 模式，因此本文基于 YARN-Cluster 模式

一、前言

1.1. 基本概念

首先说明下 Spark 里的几个概念:

一个 Spark 应用程序包括 Job、Stage 以及 Task 三个概念：

Job 是以 Action 方法为界，遇到一个 Action 方法则触发一个 Job
Stage 是 Job 的子集，以宽依赖为界。遇到 Shuffle 做一次划分
Task 是 Stage 的子集，以并行度(分区数)来衡量,分区数是多少，则有多少个 task

二、Spark 调度概述

Spark 的任务调度总体来说分两路进行，一路是 Stage 级的调度，一路是 Task 级的调度，总体调度流程如下图所示：

2.1. Stage 级调度

DAGScheduler 负责 Stage 级的调度，将 Job 以宽依赖为界切分成若干 Stages，并将每个 Stage 打包成 TaskSet 交给 TaskScheduler 调度。

DAGScheduler 做的事情较为简单，在 Stage 层面上划分 DAG，提交 Stage 并监控相关状态信息

2.2. Task 级调度

TaskScheduler 负责 Task 级的调度，将 DAGScheduler 给过来的 TaskSet 按照指定的调度策略分发到 Executor 上执行，调度过程中 SchedulerBackend 负责提供可用资源，其中 SchedulerBackend 有多种实现，分别对接不同的资源管理系统。

三、DAGScheduler

DAGScheduler 是实现了面向 stage 的调度，它可以为每个 Job 计算出一个 DAG，追踪 RDD 和 stage 的输出是否被持久化，并且寻找到一个最优调度机制来运行 Job.

接收用户提交的 Job；
将 Job 划分为不同 stage 的 DAG 图，记录哪些 RDD、Stage 被物化存储，并在每一个 stage 内产生一系列的 task，并封装成 TaskSet；
要保证相互依赖的 Job/stage 能够得到顺利的调度执行，DAGScheduler 必然需要监控当前Job / Stage乃至Task的完成情况。
结合当前的缓存情况，决定每个 Task 的最佳位置(移动计算而不是移动数据，任务在数据所在的节点上运行)，将 TaskSet 提交给 TaskScheduler;

DAGScheduler 找到哪些 RDDs 已经被 cache 了来避免重计算它们，而且同样地记住哪些ShuffleMapStages 已经生成了输出文件来避免重建一个 shuffle 的 map 侧计算任务。
重新提交 Shuffle 输出丢失的 Stage 给 TaskScheduler

处理由于 shuffle 输出文件丢失导致的失败，在这种情况下，旧的 stage 可能会被重新提交。一个 stage 内部的失败，如果不是由于 shuffle 文件丢失导致的，会被 TaskScheduler 处理，它会被多次重试每一个 task，直到最后一个。实在不行，才会被取消整个 stage。

3.1. Stage 划分

SparkContext 将 Job 提交给 DAGScheduler，DAGScheduler 将一个 Job 划分为若干 Stages ，具体划分策略是，以 Shuffle 为界，划分 Stage ,由最终的 RDD 不断通过依赖回溯判断父依赖是否是宽依赖，窄依赖的 RDD 被划分到同一个 Stage 中，进行 pipeline 式的计算，划分的 Stages 分两类，一类叫做 ResultStage 为 DAG 下游的 Stage，由 Action 方法决定；另一类叫做 ShuffleMapStage，其为下游 Stage 准备数据。

## 3.2. 生成 Job，提交 Stage

一个 Stage 是否被提交，需要判断它的父 Stage 是否执行，只有在父 Stage 执行完毕才能提交当前 Stage，如果一个 Stage 没有父 Stage，那么从该 Stage 开始提交。Stage 提交时会将 Task 信息[分区信息以及方法等]序列化并被打包成 TaskSet 交给 TaskScheduler，一个 Partition 对应一个 Task。

DAGScheduler 划分 Stage, 通过调用 submitStage 来提交一个 Stage 对应的 tasks， submitStage 通过调用 getPreferrdeLocations() 得到 task 的优先位置, 根据每个 task 的优先位置，确定 task 的 Locality 级别，Locality 一共有五种，优先级由高到低顺序

四、Task 级调度

Spark Task 的调度是由 TaskScheduler 来完成。

**TaskSetManager 负责监控管理同一个 $Stage$ 中的 $Tasks$**，TaskScheduler 会先把 DAGScheduler 给过来的 TaskSet 封装成 TaskSetManager 放到任务队列里，然后再按照指定的调度策略在调度队列中选择 TaskSetManager

4.1. 调度策略

$TaskScheduler$ 支持两种调度策略，一种是 $FIFO$，也是默认的调度策略，另一种是 FAIR。这个调度过程实际上还是比较粗粒度的，是面向 TaskSetManager 的。

在 TaskScheduler 初始化过程中会实例化 rootPool，表示树的根节点，是 Pool 类型。

FIFO 调度策略

如果是采用 FIFO 调度策略，则直接简单地将 TaskSetManager 按照先来先到的方式入队，出队时直接拿出最先进队的 TaskSetManager

private[spark] class FIFOSchedulingAlgorithm extends SchedulingAlgorithm {
  override def comparator(s1: Schedulable, s2: Schedulable): Boolean = {
    val priority1 = s1.priority // jobId
    val priority2 = s2.priority
    var res = math.signum(priority1 - priority2)
    if (res == 0) {
      val stageId1 = s1.stageId
      val stageId2 = s2.stageId
      res = math.signum(stageId1 - stageId2)
    }
    res < 0
  }
}

比较 s1和 s2 所属的 JobId，值越小，优先级越高
如果两个 JobId 的优先级相同，则对 s1，s2所属的 StageId 进行比较，值越小，优先级越高

Fair 调度策略

FAIR 模式中有一个 Root Pool 和多个子 Pool，各个子 Pool 中存储着所有待分配的 TaskSetManager 。

可以通过在 Properties 中指定 spark.scheduler.pool 属性，指定某个调度池作为 TaskSetManager 的父调度池，如果根调度池不存在此属性值对应的调度池，会创建以此属性值为名称的调度池作为 TaskSetManager 的父调度池，并将此调度池作为根调度池的子调度池。

在 FAIR 模式中，使用相同的排序算法先对子Pool 进行排序，再对子 Pool 里面的 TaskSetManager 进行排序

因为 Pool 和 TaskSetManager 都继承了 Schedulable 特质。

每个要排序的对象包含三个属性 : runningTasks 值[正在运行的 Task 数]、 minShare 值、 weight 值，比较时会综合考量三个属性值。

注意，minShare 、weight 的值均在公平调度配置文件 fairscheduler.xml 中被指定，调度池在构建阶段会读取此文件的相关配置。

<allocations>
  <pool name="production">
    <schedulingMode>FAIR</schedulingMode>
    <weight>1</weight>
    <minShare>2</minShare>
  </pool>
  <pool name="test">
    <schedulingMode>FIFO</schedulingMode>
    <weight>2</weight>
    <minShare>3</minShare>
  </pool>
</allocations>

runningTasks 比 minShare 小的先执行

如果 A 对象的 runningTasks 大于它的 minShare，B 对象的 runningTasks 小于它的 minShare,那么 B 排在 A 前面
minShare 使用率低的先执行

如果A，B 对象的 runningTasks 都小于它的 minShare ，那么就比较 runningTasks 和 minShare 的比值 [minShare使用率]谁小谁排前面
权重使用率低的先执行

如果A、B 对象的 runningTasks 都大于它们的 minShare ，那么就比较 runningTasks 与 weight 的比值(权重使用率),谁小谁排前面。

整体上来说就是通过 minShare 和 weight 这两个参数控制比较过程，可以做到

让 minShare 使用率和权重使用率少（实际运行 task 比例较少）的先运行。
如果上述比较均相等，则比较名字

从调度队列中拿到 TaskSetManager 后，由于 TaskSetManager 封装了一个 Stage 的所有 Task，并负责管理调度这些 Task，接下来 TaskSetManager 按照一定的规则逐个取出 Task 给 TaskScheduler，TaskScheduler 提交给 SchedulerBackend 去发到 Executor 执行。

2.3. 本地化调度

从调度队列中拿到TaskSetManager后，那么接下来的工作就是TaskSetManager按照一定的规则一个个取出task给TaskScheduler，TaskScheduler再交给SchedulerBackend发送到具体的Executor上执行。TaskSetManager封装了一个Stage的所有Task，并负责管理调度这些Task，每个Task都有数据依赖的优先位置，在调度执行时，Spark调度总是会尽量让每个task以最高的本地性级别来启动，当一个task以本地性级别启动，但是该本地性级别对应的所有节点都没有空闲资源而启动失败，此时并不会马上降低本地性级别启动而是在某个时间长度内再次以本地性级别来启动该task，若超过限时时间则降级启动，去尝试下一个本地性级别，依次类推。可以通过调大每个类别的最大容忍延迟时间，在等待阶段对应的Executor可能就会有相应的资源去执行此task，这就在在一定程度上提升了运行性能。

TaskSetManager 封装了一个 Stage 的所有 task，从调度队列中拿到 TaskSetManager 后，那么接下来的工作就是调度这些 task，

TaskSetManager是基于数据本地性来调度执行任务，取出 task 给TaskScheduler，TaskScheduler 再交给 SchedulerBackend 去发到Executor 上执行

TaskSetManager 根据每个 task 的优先位置，确定 task 的 Locality 级别，Locality 一共有五种，优先级由高到低顺序

PROCESS_LOCAL	进程本地化，task 和数据在同一个 Executor 中，性能最好。
NODE_LOCAL	节点本地化，task 和数据在同一个节点中，但是 task 和数据不在同一个 Executor 中，数据需要在进程间进行传输。
RACK_LOCAL	机架本地化，task 和数据在同一个机架的两个节点上，数据需要通过网络在节点之间进行传输。
NO_PREF	数据从哪里访问都一样快，不需要位置优先
ANY	task 和数据不在一个机架中，性能最差。

在调度执行时，总是会尽量让每个 $task$ 以最高的本地性级别来启动，当一个 $task$ 以 X 本地性级别启动，但是该本地性级别对应的所有节点都没有空闲资源而启动失败，此时并不会马上降低本地性级别启动, 而是在最大容忍延迟时间长度内再次以 X 本地性级别来启动该 $task$，若超过限时时间则降级启动，去尝试下一个本地性级别，依次类推。

可以通过调大每个类别的最大容忍延迟时间，在等待阶段对应的 Executor 可能就会有相应的资源去执行此 task，这就在在一定程度上提到了运行性能。

2.4. 失败重试与黑名单机制

TaskSetManager 除了调度 Task 机制外，还需要监控 Task 的执行状态。$Task$ 被调度到 $Executor$ 启动执行后，$Executor$ 会将执行状态上报给 $SchedulerBackend$， $SchedulerBackend$ 则通知该 $Task$ 对应的 $TaskSetManager$，对于失败的 $Task$，$TaskSetManager$ 会记录失败次数，如果失败次数还没有超过最大重试次数，则把该 $Task$ 放回待调度的 $Task$ 池子中，否则整个 $Application$ 失败。

在记录 $Task$ 失败次数过程中，会记录其上一次失败所在的 $ExecutorId$ 和 $Host$，下次调度该 $Task$ 时，会使用黑名单机制，避免再次被调度到上一次失败的节点上，起到一定的容错作用。

黑名单记录 $Task$ 上一次失败所在的 $ExecutorId$ 和 $Host$，以及其对应的 “拉黑时间”., “拉黑时间”是指这段时间内不要再往这个节点上调度这个 $Task$ 了。

五、总结

Spark RDD 通过其 Transactions 操作，形成了 RDD 血缘关系图，即 DAG，最后通过 Action 的调用，触发 Job 并调度执行。DAGScheduler 负责 Stage 级的调度，主要是将 DAG 切分成若干 Stages，并将每个 Stage 打包成 TaskSet 交给 TaskScheduler调度。TaskScheduler 负责 Task 级的调度，将 DAGScheduler 给过来的 TaskSet 按照指定的调度策略分发到 Executor 上执行，调度过程中 SchedulerBackend 负责提供可用资源，其中 SchedulerBackend 有多种实现，分别对接不同的资源管理系统

DAGScheduler 将 Stage 打包到 TaskSet 交给 TaskScheduler， TaskScheduler 会将 TaskSet 封装为TaskSetManager 加入到调度队列中

TaskSetManager 负责监控管理同一个 Stage 中的 Tasks，以 TaskSetManager 为单元来调度任务。