Spark 的 job、stage 和 task 的机制论述

​

Spark 任务调度机制论述

在生产环境下，Spark 集群的部署方式一般为 YARN-Cluster 模式。 Driver 线程主要是初始化 SparkContext 对象，准备运行所需的上下文，然后一方面保持与 ApplicationMaster 的 RPC 连接，通过 ApplicationMaster 申请资源，另一方面根据用户业务逻辑开始调度任务，将任务下发到已有的空闲 Executor 上。当 ResourceManager 向 ApplicationMaster 返回 Container 资源时，ApplicationMaster 就尝试在对应的 Container 上启动 Executor 进程，Executor 进程起来后，会向 Driver 反向注册，注册成功后保持与 Driver 的心跳，同时等待 Driver 分发任务，当分发的任务执行完毕后，将任务状态上报给 Driver。

1. Spark 任务调度概述

1.1 基础概念

当 Driver 起来后，Driver 则会根据用户程序逻辑准备任务，并根据 Executor 资源情况逐步分发任务。在详细阐述任务调度前，首先说明下 Spark 里的几个概念。一个 Spark 应用程序包括 Job、Stage 以及 Task 三个概念：

job：以 action 方法为界，一个 action 触发一个 job

stage：它是 job 的子集，以 RDD 宽依赖为界，遇到宽依赖即划分 stage

task：它是 stage 的子集，以分区数来衡量，分区数多少，task 就有多少

​

1.2 任务调度

spark 任务从发起到执行可用下图表示

​

1.3 Client—>ResourceManage

(1). Client 端通过 spark-submit + 参数 发起任务，即向 ResourceManage 提交 application，注意该 application 包含了一堆参数，如 Executor 数，Executor 内存，Driver 内存等；

(2). ResourceManage 需要先判断现在资源是否能满足该 application，如果满足，则响应该 application，如果不满足，报错；

(3). 如果资源满足，Client 端准备 ApplicationMaster 的启动上下文，并交给 ResourceManage；

(4). 并且循环监控 application 的状态；

1.4 ResourceManage—>ApplicationMaster

(1). ResourceManage 找一个 worker 启动 ApplicationMaster；

(2). ApplicationMaster 向 ResourceManage 申请 Container；

(3). ResourceManage 收集可用资源，并告诉 ApplicationMaster；

(4). ApplicationMaster 尝试在对应的 Container 上启动 Executor 进程；

1.5 ApplicationMaster-Driver

(1). 有了资源，ApplicationMaster 启动 Driver；

//Driver 线程主要是初始化 SparkContext 对象，准备运行所需上下文，并保持与 ApplicationMaster 的 RPC 连接，通过 ApplicationMaster 申请资源

(2). Driver 启动成功后，告诉 ApplicationMaster；

1.6 Driver-Executor

(1). Executor 启动成功后，反向注册到 Driver 上，并持续向 Driver 发送心跳；

(2). Driver 启动 task，分发给 Executor，并监控 task 状态；

(3). 当 Executor 任务执行完毕后，将任务状态发送给 Driver；

spark 的核心就是资源申请和任务调度，主要通过 ApplicationMaster、Driver、Executor 来完成

spark 任务调度分为两层，一层是 stage 级的调度，一层是 task 级的调度

​

RDD 间的血缘关系，代表了计算的流程，构成了 有向无环图，即 DAG；

最后通过 action 触发 job 并调度执行；

DAGScheduler 负责 stage 级的调度，主要是将 DAG 切分成多个 stage，并将 stage 打包成 TaskSet 交给 TaskScheduler；

TaskScheduler 负责 task 级的调度，将 DAGScheduler 发过来的 TaskSet 按照指定的调度策略发送给 Executor；

SchedulerBackend 负责给 调度策略 提供可用资源，调度策略决定把 task 发送给哪个 Executor；【其中 SchedulerBackend 有多种实现，分别对接不同的资源管理系统】

基于上述认知，再来看一张图

​

Driver 在启动过程中，除了初始化 SparkContext 外，也初始化了 DAGScheduler、TaskScheduler、 SchedulerBackend 3 个调度对象，同时初始化了 HeartbeatReceiver 心跳接收器；

并且各个线程之间保存通信；

SchedulerBackend 向 ApplicationMaster 申请资源，并不间断地从 TaskScheduler 获取 task 并发送给 合适的 Executor；

HeartbeatReceiver 负责接收 Executor 心跳报文，监控 Executor 存活状态；

2. Spark Stage 级调度

Spark 的任务调度是从 DAG 切割开始，主要是由 DAGScheduler 来完成。当遇到一个 Action 操作后就会触发一个 Job 的计算，并交给 DAGScheduler 来提交，下图是涉及到 Job 提交的相关方法调用流程图。

​

1) Job 由最终的 RDD 和 Action 方法封装而成；

2) SparkContext 将 Job 交给 DAGScheduler 提交，它会根据 RDD 的血缘关系构成的 DAG 进行切分，将一个 Job 划分为若干 Stages，具体划分策略是，由最终的 RDD 不断通过依赖回溯判断父依赖是否是宽依赖，即以 Shuffle 为界，划分 Stage，窄依赖的 RDD 之间被划分到同一个 Stage 中，可以进行 pipeline 式的计算。划分的 Stages 分两类，一类叫做 ResultStage，为 DAG 最下游的 Stage，由 Action 方法决定，另一类叫做 ShuffleMapStage，为下游 Stage 准备数据，下面看一个简单的例子 WordCount。

​

Job 由 saveAsTextFile 触发，该 Job 由 RDD-3 和 saveAsTextFile 方法组成，根据 RDD 之间的依赖关系从 RDD-3 开始回溯搜索，直到没有依赖的 RDD-0，在回溯搜索过程中，RDD-3 依赖 RDD-2，并且是宽依赖，所以在 RDD-2 和 RDD-3 之间划分 Stage，RDD-3 被划到最后一个 Stage，即 ResultStage 中，RDD-2 依赖 RDD-1，RDD-1 依赖 RDD-0，这些依赖都是窄依赖，所以将 RDD-0、RDD-1 和 RDD-2 划分到同一个 Stage，形成 pipeline 操作，。即 ShuffleMapStage 中，实际执行的时候，数据记录会一气呵成地执行 RDD-0 到 RDD-2 的转化。不难看出，其本质上是一个深度优先搜索（Depth First Search）算法。

一个 Stage 是否被提交，需要判断它的父 Stage 是否执行，只有在父 Stage 执行完毕才能提交当前 Stage，如果一个 Stage 没有父 Stage，那么从该 Stage 开始提交。Stage 提交时会将 Task 信息（分区信息以及方法等）序列化并被打包成 TaskSet 交给 TaskScheduler，一个 Partition 对应一个 Task，另一方面 TaskScheduler 会监控 Stage 的运行状态，只有 Executor 丢失或者 Task 由于 Fetch 失败才需要重新提交失败的 Stage 以调度运行失败的任务，其他类型的 Task 失败会在 TaskScheduler 的调度过程中重试。

相对来说 DAGScheduler 做的事情较为简单，仅仅是在 Stage 层面上划分 DAG，提交 Stage 并监控相关状态信息。TaskScheduler 则相对较为复杂，下面详细阐述其细节。

3. Spark Task 级调度

Spark Task 的调度是由 TaskScheduler 来完成，由前文可知，DAGScheduler 将 Stage 打包到交给 TaskScheTaskSetduler，TaskScheduler 会将 TaskSet 封装为 TaskSetManager 加入到调度队列中，TaskSetManager 结构如下图所示。

​

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

前面也提到，TaskScheduler 初始化后会启动 SchedulerBackend，它负责跟外界打交道，接收 Executor 的注册信息，并维护 Executor 的状态，所以说 SchedulerBackend 是管“粮食”的，同时它在启动后会定期地去“询问”TaskScheduler 有没有任务要运行，也就是说，它会定期地“问”TaskScheduler“我有这么余粮，你要不要啊”，TaskScheduler 在 SchedulerBackend“问”它的时候，会从调度队列中按照指定的调度策略选择 TaskSetManager 去调度运行，大致方法调用流程如下图所示：

​

上图中，将 TaskSetManager 加入 rootPool 调度池中之后，调用 SchedulerBackend 的 riviveOffers 方法给 driverEndpoint 发送 ReviveOffer 消息；driverEndpoint 收到 ReviveOffer 消息后调用 makeOffers 方法，过滤出活跃状态的 Executor（这些 Executor 都是任务启动时反向注册到 Driver 的 Executor），然后将 Executor 封装成 WorkerOffer 对象；准备好计算资源（WorkerOffer）后，taskScheduler 基于这些资源调用 resourceOffer 在 Executor 上分配 task。

3.1 调度策略

TaskScheduler 支持两种调度策略，一种是 FIFO，也是默认的调度策略，另一种是 FAIR。在 TaskScheduler 初始化过程中会实例化 rootPool，表示树的根节点，是 Pool 类型。

(1) FIFO 调度策略

如果是采用 FIFO 调度策略，则直接简单地将 TaskSetManager 按照先来先到的方式入队，出队时直接拿出最先进队的 TaskSetManager，其树结构如下图所示，TaskSetManager 保存在一个 FIFO 队列中。

​

(2) FAIR 调度策略

FAIR 调度策略的树结构如下图所示：

​

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

在 FAIR 模式中，需要先对子 Pool 进行排序，再对子 Pool 里面的 TaskSetMagager 进行排序，因为 Pool 和 TaskSetMagager 都继承了 Schedulable 特质，因此使用相同的排序算法。

排序过程的比较是基于 Fair-share 来比较的，每个要排序的对象包含三个属性: runningTasks 值（正在运行的 Task 数）、minShare 值、weight 值，比较时会综合考量 runningTasks 值，minShare 值以及 weight 值。

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

1) 如果 A 对象的 runningTasks 大于它的 minShare，B 对象的 runningTasks 小于它的 minShare，那么 B 排在 A 前面；（runningTasks 比 minShare 小的先执行）

2) 如果 A、B 对象的 runningTasks 都小于它们的 minShare，那么就比较 runningTasks 与 minShare 的比值（minShare 使用率），谁小谁排前面；（minShare 使用率低的先执行）

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

4) 如果上述比较均相等，则比较名字。

整体上来说就是通过 minShare 和 weight 这两个参数控制比较过程，可以做到让 minShare 使用率和权重使用率少（实际运行 task 比例较少）的先运行。

FAIR 模式排序完成后，所有的 TaskSetManager 被放入一个 ArrayBuffer 里，之后依次被取出并发送给 Executor 执行。

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

3.2 本地化调度

DAGScheduler 切割 Job，划分 Stage, 通过调用 submitStage 来提交一个 Stage 对应的 tasks，submitStage 会调用 submitMissingTasks，submitMissingTasks 确定每个需要计算的 task 的 preferredLocations，通过调用 getPreferrdeLocations()得到 partition 的优先位置，由于一个 partition 对应一个 Task，此 partition 的优先位置就是 task 的优先位置，对于要提交到 TaskScheduler 的 TaskSet 中的每一个 Task，该 task 优先位置与其对应的 partition 对应的优先位置一致。

从调度队列中拿到 TaskSetManager 后，那么接下来的工作就是 TaskSetManager 按照一定的规则一个个取出 task 给 TaskScheduler，TaskScheduler 再交给 SchedulerBackend 去发到 Executor 上执行。前面也提到，TaskSetManager 封装了一个 Stage 的所有 Task，并负责管理调度这些 Task。

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

​

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

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

3.3 失败重试与黑名单机制

除了选择合适的 Task 调度运行外，还需要监控 Task 的执行状态，前面也提到，与外部打交道的是 SchedulerBackend，Task 被提交到 Executor 启动执行后，Executor 会将执行状态上报给 SchedulerBackend，SchedulerBackend 则告诉 TaskScheduler，TaskScheduler 找到该 Task 对应的 TaskSetManager，并通知到该 TaskSetManager，这样 TaskSetManager 就知道 Task 的失败与成功状态，对于失败的 Task，会记录它失败的次数，如果失败次数还没有超过最大重试次数，那么就把它放回待调度的 Task 池子中，否则整个 Application 失败。

在记录 Task 失败次数过程中，会记录它上一次失败所在的 Executor Id 和 Host，这样下次再调度这个 Task 时，会使用黑名单机制，避免它被调度到上一次失败的节点上，起到一定的容错作用。黑名单记录 Task 上一次失败所在的 Executor Id 和 Host，以及其对应的“拉黑”时间，“拉黑”时间是指这段时间内不要再往这个节点上调度这个 Task 了。

4. 总结

本图有助于理解 job，stage，task 工作的原理。Spark 通用运行流程图，体现了基本的 Spark 应用程序在部署中的基本提交流程。

​

流程按照如下的核心步骤进行工作的：

1) 任务提交后，都会先启动 Driver 程序；

2) 随后 Driver 向集群管理器注册应用程序；

3) 之后集群管理器根据此任务的配置文件分配 Executor 并启动；

4) Driver 开始执行 main 函数，Spark 查询为懒执行，当执行到 Action 算子时开始反向推算，根据宽依赖进行 Stage 的划分，随后每一个 Stage 对应一个 Taskset，Taskset 中有多个 Task，查找可用资源 Executor 进行调度；

5) 根据本地化原则，Task 会被分发到指定的 Executor 去执行，在任务执行的过程中，Executor 也会不断与 Driver 进行通信，报告任务运行情况。

​

关键词：大数据培训