7-1、Spark-Scheduler

分手后的思念是犯贱 2022-08-21 03:10 144阅读 0赞

# 4、Scheduler #

Scheduler（任务调度）模块是Spark Core的核心模块之一。

Spark对于DAG(Directed Acyclic Graph，有向无环图)的实现以及不同执行阶段的划分和任务的提交执行。

任务调度，即组成应用的多个Job之间如何分配计算资源。

### 4.1、整体模块概述 ###

4.1.1、DAGScheduler和TaskScheduler

任务调度模块主要包含两大部分：DAGScheduler和TaskScheduler，它们负责将用户提交的计算任务按照DAG划分为不同的阶段并且将不同阶段的计算任务提交到集群进行最终的计算。

下图为任务调度逻辑视图：

![Center][]

RDD Objects可以理解为用户实际代码中创建的RDD，这些代码逻辑上组成一个DAG。用户实际编程的时候可以认为它处理的数据都是可以存在内存中，而无需关心最终在集群中运行的任务是否整个数据都可以装载到内存中或者说它究竟需要多少个节点参与运算。

DAGScheduler主要负责分析用户提交的应用，并根据任务的依赖关系建立DAG，然后将DAG划分为不同的Stage（阶段），其中每个Stage由可以并发执行的一组Task构成，这些Task的执行逻辑完全相同，只是作用不同的数据。而且DAG在不同的资源管理框架下的实现是相同的。

DAGScheduler将这组Task划分完成后，会将这组Task提交到TaskScheduler。TaskScheduler通过Cluster Manager在集群中的某个Worker的Executor上启动任务。

用户提交的Job最终会调用DAGScheduler的runJob，它又会调用submitJob。

Submit首先会为这个Job生成一个Job ID，并且生成一个JobWaiter的实例来监听Job的执行状态。

JobWriter会监听Job的执行状态，而Job是由多个Task组成的，因此只有Job的所有Task都成功完成，Job才标记为成功，任意一个Task失败都会标记这个Job失败。

### 4.2、Job的Stage的划分 ###

用户提交的计算任务是由一个个RDD构成的DAG，如果RDD在转换的时候需要做Shuffle，那么这个Shuffle的过程，就将这个DAG分为了不同的阶段（即Stage）。

由于Shuffle的存在，不同的Stage是不能并行计算的，因为后面的Stage的计算需要前面的Stage的Shuffle的结果。

一个Stage由一组完全独立的计算任务（即Task组成），每个Task的运行逻辑完全相同，只不过每个Task都会处理其所对应的Partition。如果Partition的数量和Task数量一致，那么一个Partition会被该Stage的一个Task处理。

Spark根据宽依赖将Job划分为不同的阶段（Stage）。

下图为RDD划分示意图：

![Center 1][]

Stage的划分是从最后一个RDD开始的，也就是触发Action的那个RDD。

根据上图分析：

划分从RDD G的依赖开始：

RDD G依赖两个RDD，RDD B和RDD F，其中先处理哪一个是随机的。假如先处理RDD B，由于RDD B和RDD G是窄依赖，因此RDD B和RDD G可以划分到一个Stage（即Stage3）。 再处理RDD F，因为这个依赖是宽依赖，所以RDD F和RDD G被划分到不同的Stage（即Stage 2和Stage3），其中RDD F所在的Stage2是RDD G所在的Stage3的parent Stage。

接着处理RDD B的依赖，由于它依赖的RDD A是宽依赖，因此它们属于不同的Stage（Stage1和Stage3）

最后，这个DAG被划分为3个Stage，即RDD A所在的Stage1，RDD C、RDD D、RDD E、RDD F所在的Stage2和RDD B、RDD G所在的Stage3。

其中Stage1和Stage2是相互独立的，可以并行执行。Stage3依赖于Stage1和Stage2，只有Stage1和Stage2计算完成之后，它才可以开始计算。

### 4.3、应用Application ###

TaskScheduler和DAGScheduler都是在SparkContext创建的时候创建的。DAGScheduler保存了TaskScheduler的引用，因此需要在TaskScheduler创建之后进行创建。

TaskScheduler负责Application的不同Job之间的调度，在Task执行失败的时候启动重试机制，并且为执行速度慢的Task启动备份的任务。SchedulerBackend负责与Cluster Manager交互，取得该Application分配到的资源，并将这些资源传给TaskScheduler，由TaskScheduler为Task最终分配计算资源。

![Center 2][]

应用的基本组件：

Application：用户自定义的Spark程序，用户提交后，Spark为App分配资源，将程序转换并执行。

Driver Program：运行Application的main()函数并创建SparkContext。

RDD Graph：RDD是Sparkde 核心结构，可以通过一系列的算子进行操作。当RDD遇到Action算子时，将之前的所有算子形成一个有向无环图（DAG），也就是图中的RDD Graph。再在Spark中转化为Job，提交到集群执行。一个App中可以包含多个Job。

Job：一个RDD Graph触发的作业，往往由Spark Action算子触发，在SparkContext中通过runJob方法向Spark提交Job。

Stage：每个Job会根据RDD的宽依赖关系被切分成很多Stage，每个Stage中包含一组相同的Task，这一组Task叫TaskSet。

Task：一个分区对应一个Task，Task执行RDD 中对应Stage中包含的算子。Task被封装好后放入Executor的线程池中执行。

Driver进程是应用的主控进程，负责应用的解析、切分并调度Task到Executor执行，包含DAGScgeduler等重要对象。

Master调度应用，针对每一个应用分发给指定的一个Worker启动Driver，即SchedulerBackend。SchedulerBackend进程中包含DAGScheduler，它会根据RDD的DAG切分Stage，生成TaskSet，并调度和分发Task到Executor。

### 4.4、任务调度实现 ###

调度分为4个级别：Application调度、Job调度、Stage调度、Task的调度与分发。

4.4.1、Application调度：

SparkContext维持着整个应用的上下文，提供一些核心方法。如runJob提交Job，然后通过主节点的分配获得一组Executor JVM进程执行任务。Executor空间内的不同应用之间是不共享的，一个Executor在一个时间段内只能分配给一个应用使用。

默认情况下，用户向以Standalone模式运行的Spark集群提交的应用使用FIFO（先进先出）的顺序进行调度，每个应用会独占所有可用节点的资源。

在Standalone、Mesos和YARN模式都不提供跨应用的共享内存。

4.4.2、Job调度：

在Spark应用程序内部，用户通过不同线程提交的Job可以并行运行，这里的Job就是Spark Action算子触发的整个RDD DAG为一个Job。实现上就是调用runJob方法提交的Job。

下图为FIFO调度模式：

![Center 3][]

在算法执行中，先看优先级，TaskSet的优先级是JobID，因为先提交的Job ID小，所以就会被更优先的调度。这里相当于进行了两层排序，先看是否是同一个Job的TaskSet，不同Job之间的TaskSet先排序。

最后执行的satageID最小为0，最先应该执行的stageID最大。但是这里的调度机制是优先调度StageID小的。在DAGScheduler中控制Stage是否被提交到队列，如果还有父母Stage未执行完，则该stage的Taskset不会被提交到调度池中。这就保证了虽然最先执行的stage的id最大，但是排序完，由于后面的还没有提交到调度池，所以会先执行。

Stage的TaskSet调度逻辑主要在DAGScheduler中，而Job调度由FIFO或者FAIR算法调度。

下图为FAIR调度模型：

![Center 4][]

自Spark0.8版本后，可以开始通过配置FAIR共享调度模式调度Job。

在FAIR共享模式调度下，Spark在多Job之间以轮训方式为任务分配资源，所有任务拥有大致相当的优先级来共享集群的资源。

FAIR调度器同样支持将Job分组加入调度池调度，用户可以同时针对不同优先级对每个调度池配置不同的调度权重。这种方式允许更重要的Job配置在高优先级池中优先调度。这种方式借鉴了Hadoop的FAIR调度器。

配置调度池：

1）、调度模式：用户可以选择FIFO或FAIR

2）、权重：作用在控制整个集群资源分配上，这个调度池相对其他调度池优先级的高低。

3）、minShare：这个参数决定整体调度的调度池能给待调度的调度池分配多少资源就可以满足调度池的资源需求，剩下的资源可以分配给其他调度池。

4.4.3、Stage调度：

Stage的调度由DAGScheduler完成的。由RDD 的有向无环图DAG切分出了Stage的有向无环图DAG。

Stage的DAG通过最后执行的Stage为根进行广度优先遍历，便利到最开始执行的Stage执行，如果提交的stage仍有未完成的父母stage，则Stage需要等待其父母执行完才能执行。

在TaskScheduler中将每个Stage中对应的任务进行提交和调度。

一个应用对应一个TaskScheduler，也就是这个应用中所有Action触发的Job中的TaskSetManager都是由这个TaskScheduler调度的。

每个Stage对应的一个TaskSetManager通过Stage回溯到最源头缺失的Stage提交到调度池pool中，在调度池中，这些TaskSetManager又会根据Job ID排序，先提交的Job的TaskSetManager优先调度，然后一个Job内的TaskSetManager ID小的先调度，并且如果有未执行完的父母Stage的TaskSetManager，则不会提交到调度池中。

4.4.4、Task调度

整体的Task分发由TaskSchedulerImpl来实现，但是Task的调度（本质上是Task在哪个分区上执行）逻辑由TaskSetManager完成。这个类监控这个任务的生命周期，当任务失败时，重新调度。

如果是调用过cache()方法的RDD，数据已经缓存在内存，则读取内存缓存中分区的数据。

如果能直接获取执行地点，则返回执行地点作为任务的执行地点。通常DAG中最源头的RDD或者每个Stage中最开始的RDD会有执行地点的信息。例如HadoopRDD从HDFS读取的分区就是最好的执行地点。

如果以上2种情况都不是，则将遍历RDD获取第一个窄依赖的父RDD对应分区的执行地点。

[Center]: /images/20220731/b8ab6d62b27c4f0bb75c6c5caab4ab0c.png
[Center 1]: /images/20220731/24e036f6bf004dba905f8527dd58c2a5.png
[Center 2]: /images/20220731/461a9213d6f14ff6b6687bb8b35341a3.png
[Center 3]: /images/20220731/a345e33b2a814f518948762009e6949f.png
[Center 4]: /images/20220731/994c55733ae44fb29ec01d4649b26789.png