Spark-TaskSchedule和TaskScheduleImpl解释和过程

系统管理员 2024-02-19 15:14 30阅读 0赞

### 一：什么是TaskSchedule ###

\*\*  
**官网的解释  
Low-level task scheduler interface, currently implemented exclusively by TaskSchedulerImpl.This interface allows plugging in different task schedulers. Each TaskScheduler schedules tasks for a single SparkContext. These schedulers get sets of tasks submitted to them from the DAGScheduler for each stage, and are responsible for sending the tasks to the cluster, running them, retrying if there are failures, and mitigating stragglers. They return events to the DAGScheduler.**  
\*\*  
TaskSchedule是一个低层次的任务调度接口，目前只有TaskScheduleImpl实现了它，这个接口允许使用不同的任务调度策略。每一个任务调度器只服务于一个SparkContext。TaskSchedule会从DAGSchedule那边获取每一个stage的tasks的集合，并且会负责将它们提交到集群上去运行，还会在任务失败的时候重新提交它们

### 二：什么是TaskScheduleImpl ###

**官网的解释  
Schedules tasks for multiple types of clusters by acting through a SchedulerBackend.  
It can also work with a local setup by using a LocalBackend and setting isLocal to true.  
It handles common logic, like determining a scheduling order across jobs, waking up to launch  
speculative tasks, etc.  
Clients should first call initialize() and start(), then submit task sets through the  
runTasks method.  
TaskSchedule**  
ScheduleBackend来调度不同类型的集群的任务  
可以使用LocalBackend来处理local集群的任务  
处理一些普通的逻辑，比如确定job之间的调度顺序，唤醒一些预测的任务  
应该首先调用initialize和start方法，然后通过runTasks方法提交任务集  
TaskScheduler与SchedulerBackend

### 三：总体的底层任务调度的过程如下： ###

a>TaskSchedulerImpl.submitTasks主要的作用是将TaskSet加入到TaskSetManager中进行管理；

b>SchedulableBuilder.addTaskSetManager: SchedulableBuilder会确定TaskSetManager的调度顺序，然后按照TaskSetManager的locality aware来确定每个Task 具体运行在哪个ExecutorBackend中。

c>CoarseGrainedSchedulerBackend.reviveOffers:给DriverEndpoint发送ReviveOffouers，ReviveOffouers本身是一个空的case object对象，只是起到触发底层资源调度的作用，在有Task提交或计算资源变动的是时候会发送ReviveOffers作为触发器。

d>在DriverEndpoint接受ReviveOffouers并路由到makeOffers具体的方法中，在makeOffers中，首先准备好所有可以用于计算的workOffers（代表了所有可用的ExecutorBackend中可以使用的Cores信息）

e>TaskSchedulerImpl.resourceOffers为每一个Task具体分配计算资源，输入是ExecutorBackend可用的cores，输出是TaskDescription的二维数组，在其中确定了每个Task具体运行在哪个ExecutorBackend；resourceOffers到底是如何确定Task具体运行在哪个ExecutorBackend上的?

i.通过Random.shuffle方法重新洗牌所有的计算资源以寻求计算的负载均衡

ii.根据每个ExecutorBackend的cores的个数声明类型为TaskDescription的ArrayBuffer数组

iii.如果有新的ExecutorBackend分配给我们的Job，此时会调用executorAdded来获得最新的完整的可用计算资源

iv.通过调用TaskSetManager的resourceOffers最终确定每个Task具体运行在哪个ExecutorBackend的具体的Locality Level

f>通过launchTasks把任务发送给ExecutorBackend去执行

### 四：解释 ###

（1）Task默认的最大重试次数是4次。

（2）Spark应用程序目前支持两种调度器:FIFO、FAIR，可以通过spark-env.sh中的spark.scheduler.mode进行具体的设置；默认情况下是FIFO的模式。

（3）TaskScheduler中要负责为Task分配资源：根据计算本地性原则确定Task具体要运行在哪个ExecutorBankend中，此时程序已经具备集群中的计算资源了。

（4）TaskDescription中已经确定好了Task具体要运行在哪个ExecutorBankend中，而确定Task具体要运行在哪个ExecutorBankend中的算法是由TaskSetManager的resourceOffers决定的。

（5）数据本地优先级从高到低依次为：PROCESS\_LOCAL, NODE\_LOCAL, NO\_PREF, RACK\_LOCAL, ANY, 其中NO\_PREF是指机器本地性。

（6）每个Task默认是采用一个线程进行计算的。