Spark源码分析之Scheduler模块(TaskScheduler)

DAGScheduler将任务提交到TaskScheduler之后，接下来由TaskScheduler负责任务的调度。

TaskScheduler是一个trait（接口类），它的实现类是TaskSchedulerImpl。具体内容包括：
1、出现shuffle输出lost要报告fetch failed错误
2、碰到straggle任务需要放到别的节点上重试
3、为每个TaskSet维护一个TaskSetManager
TaskScheduler与其他类之间的关系如下图中所示：



SparkContext初始化的同时会创建TaskScheduler和DAGScheduler。也就是说TaskScheduler为某个特定的SparkContext调度task。

SparkContext创建过程中会调用createTaskScheduler函数来启动TaskScheduler任务调度器：



createTaskScheduler函数中，TaskScheduler会根据部署方式而选择不同的SchedulerBackend来处理。针对不同的部署方式会有不同的TaskScheduler与SchedulerBackend进行组合：
     Local模式：TaskSchedulerImpl＋LocalBackend
     Spark集群模式：TaskSchedulerImpl＋SparkDepolySchedulerBackend
     Yarn-cluster模式：YarnClusterScheduler＋CoarseGrainedSchedulerBackend
     Yarn-Client模式：YarnClientClusterScheduler＋YarnClientSchedulerBackend

SchedulerBackend是Spark中一个可插拔组件。按照字面意思，它就是调度器的一个后台服务或者实现，其主要作用就是在物理机器或者说worker就绪后，能够提供其上的资源并将tasks加载到那些机器或者worker上。

以Standalone模式为例，backend根据不同的部署方式实例化，后又作为scheduler对象的一个成员变量对scheduler调用initialize函数：



TaskScheduler、TaskSchedulerImpl、SchedulerBackend之间的关系：
TaskScheduler类负责任务调度资源的分配，SchedulerBackend负责与不同的资源调度系统适配，与YARN/Mesos等集群中的Worker节点通信，收集Worker上分配给该应用使用的资源情况。TaskSchedulerImpl接收DAGScheduler中提交上来的TaskSet，并将TaskSet中的Task下发到集群中的计算节点去运行。



在TaskSchedulerImpl的start方法中实际上调用backend的start（backend在TaskScheduler的initiallize方法中作为参数传入的），不同的backend适配不同的资源调度方式，以sparkDeploySchedulerBackend为例，依次讲述TaskSchedulerImpl的启动、任务提交与停止：

TaskSchedulerImpl的启动：

TaskSchedulerImpl的启动方法一次调用了SparkDeploySchedulerBackend的启动方法&AppClient的启动方法，AppClient的start方法调用如下图中的所示，注意的是SparkDeploySchedulerBackend通过super.start()启动了父类CoarseGrainedSchedulerBackend。



在AppClient内部有一个RpcEndpointRef类型的变量，而start方法就是通过该变量发送一个endpoint。

TaskScheduler的任务提交：

在DAGScheduler一节中，我们讲到DAGScheduler是通过调用了TaskSchedulerImpl的submitTasks方法用于向集群中提交taskSet，进入TaskSchdulerImpl的submitTasks方法，可以观察到其内部启用了一个定时器，如果任务一直未被执行，定时器会持续发出警告直到任务被执行。

任务执行是调用了backend的reviveOffers()方法，该方法会通过actor向driver发送ReviveOffers，driver收到ReviveOffers后调用makeOffers()。submitTasks每提交一组任务的同时，会针对改组任务new一个TaskSetManager对象，TaskSetManager是TaskSet的管理者，主要用于在TaskSchedulerImpl中调度同一个TaskSet中的tasks，该类追踪每一个task，当任务失败时重试，并通过延迟调度处理位置敏感调度。该类最主要的接口resourceOffer()，该方法会询问TaskSet，它是否想要在一个节点上运行一个task。



TaskScheduler的资源申请（resourceOffers）：

前面提到的任务提交会向driver传递一个ReviveOffers对象，driver收到ReviveOffers后调用makeOffers()方法。（父类CoarseGrainedSchedulerBackend中提供了该方法的处理），在makeoffers中调用了resourceOffers方法向scheduler申请资源，并向executor提出launchTask请求。代码如下图中所示：



resourceOffers首先处理新的executor加入（调用DAGScheduler&TaskSetManager的executorAdded），接着将Task随机打散，使Task均匀分布在各Worker节点上，组建Task的执行队列（调用rootPool.getSortedTaskSetQueue），最后根据就近原则选择Task调度执行（调用resourceOfferSingleTaskSet，上述方法再进一步调用了TaskSetManager的resourceOffer方法），整个的调用流程图如下图中所示：



接下来lauchTask会进入executor模块，StandaloneExecutorBackend在收到launchTask请求后会调用Executor执行task。Executor内部是一个线程池，每一个提交的task都会被包装为TaskRunner交由threadpool执行。

thread pool中的task.run()真正执行了task中的任务，计算的返回值被包装成TaskResult返回。

任务执行结果的处理：    

statusUpdate跟踪各任务的执行状态，并根据任务的不同状态进行不同的处理。主要逻辑在下面的代码中：



失败&成功的任务分别进入enqueueFailedTask&enqueueSuccessfulTask中。
 
以enqueueSuccessfulTask为例分析，它是TaskResultGetter中定义的方法，用于处理成功的任务，在内部逻辑中，其调用了scheduler的handleSuccessfulTask方法，层层调用，最终在TaskSetManager中调用handleSuccessfulTask将该任务的相关信息移除，后半部分移除的逻辑在TaskSchedulerImpl的taskSetFinished中实现。

调度模式的选择（schedule mode）：

调度模式的确定在TaskSchedulerImpl的初始化函数中实现，具体可以参见schedulableBuilder。

schedulableBuilder随着TaskSetManager一起创建，schedulableBuilder充当调度构造器的角色，其管理着一个TaskPool，并提供了FIFO&FAIR两种调度模式，用于调度TaskPool中存在的task。它包括两个主要方法，分别是：
1、buildPools方法：构造调度树节点；
2、addTaskSetManager：构造叶子节点（TaskSetManager）

注：本文的多数图片来自https://github.com/jaceklaskowski/mastering-apache-spark-book/，谢谢这位外国网友，欢迎与大家一起学习spark相关技术。

参考资料：
1、http://jerryshao.me/architecture/2013/04/21/Spark源码分析之-scheduler模块/
2、http://blog.csdn.net/anzhsoft/article/details/39762587
3、http://blog.csdn.net/lovehuangjiaju/article/category/5640777/2