spark internal - 做业调度

在Spark中做业调度的相关类最重要的就是DAGScheduler，DAGScheduler顾名思义就是基于DAG图的Scheduler

DAG全称 Directed Acyclic Graph，有向无环图。简单的来讲，就是一个由顶点和有方向性的边构成的图中，从任意一个顶点出发，没有任何一条路径会将其带回到出发的顶点。

在做业调度系统中，调度的基础就在于判断多个做业任务的依赖关系，这些任务之间可能存在多重的依赖关系，也就是说有些任务必须先得到执行，而后另外的相关依赖任务才能执行，可是任务之间显然不该该出现任何直接或间接的循环依赖关系，因此本质上这种关系适合用DAG有向无环图来表示。

归纳地描述DAGScheduler和TaskScheduler（关于TaskScheduler的相关细节，在我以前的关于Spark运行模式的文章中有）的功能划分就是：TaskScheduler负责实际每一个具体任务的物理调度，DAGScheduler负责将做业拆分红不一样阶段的具备依赖关系的多批任务，能够理解为DAGScheduler负责任务的逻辑调度。

基本概念

Task任务 ：单个分区数据集上的最小处理流程单元

TaskSet任务集：一组关联的，可是互相之间没有Shuffle依赖关系的任务所组成的任务集

Stage调度阶段：一个任务集所对应的调度阶段

Job做业：一次RDD Action生成的一个或多个Stage所组成的一次计算做业

运行方式

DAGScheduler在SparkContext初始化过程当中实例化，一个SparkContext对应一个DAGScheduler，DAGScheduler的事件循环逻辑基于Akka  Actor的消息传递机制来构建，在DAGScheduler的Start函数中建立了一个eventProcessActor用来处理各类DAGSchedulerEvent，这些事件包括做业的提交，任务状态的变化，监控等等

private[scheduler]case class JobSubmitted(
    jobId: Int,
    finalRDD: RDD[_],
    func: (TaskContext, Iterator[_]) => _,
    partitions: Array[Int],
    allowLocal: Boolean,
    callSite: String,
    listener: JobListener,
    properties: Properties = null)
  extends DAGSchedulerEvent
 
private[scheduler]case class JobCancelled(jobId: Int) extends DAGSchedulerEvent
private[scheduler]case class JobGroupCancelled(groupId: String) extends DAGSchedulerEvent
private[scheduler]case object AllJobsCancelled extends DAGSchedulerEvent
private[scheduler]
case classBeginEvent(task: Task[_], taskInfo: TaskInfo) extends DAGSchedulerEvent
 
private[scheduler]
case classGettingResultEvent(task: Task[_], taskInfo: TaskInfo) extends DAGSchedulerEvent
 
private[scheduler]case class CompletionEvent(
    task: Task[_],
    reason: TaskEndReason,
    result: Any,
    accumUpdates: Map[Long, Any],
    taskInfo: TaskInfo,
    taskMetrics: TaskMetrics)
  extends DAGSchedulerEvent
 
private[scheduler]case class ExecutorAdded(execId: String, host: String) extendsDAGSchedulerEvent
private[scheduler]case class ExecutorLost(execId: String) extends DAGSchedulerEvent
private[scheduler]  caseclass TaskSetFailed(taskSet: TaskSet, reason: String) extends DAGSchedulerEvent
private[scheduler]case object ResubmitFailedStages extends DAGSchedulerEvent
private[scheduler]case object StopDAGScheduler extends DAGSchedulerEvent

 

不管是Client仍是TaskScheduler与DAGScheduler的交互方式基本上都是经过DAGScheduler暴露的函数接口间接的给eventProcessActor发送相关消息

如前面所说，DAGScheduler最重要的任务之一就是计算做业和任务的依赖关系，制定调度逻辑

DAGScheduler做业调度的两个主要入口是submitJob 和 runJob，二者的区别在于前者返回一个Jobwaiter对象，能够用在异步调用中，用来判断做业完成或者取消做业，runJob在内部调用submitJob，阻塞等待直到做业完成（或失败）

具体往DAGScheduler提交做业的操做，基本都是封装在RDD的相关Action操做里面，不须要用户显式的提交做业

用户代码都是基于RDD的一系列计算操做，实际运行时，这些计算操做是Lazy执行的，并非全部的RDD操做都会触发Spark往Cluster上提交实际做业，基本上只有一些须要返回数据或者向外部输出的操做才会触发实际计算工做，其它的变换操做基本上只是生成对应的RDD记录依赖关系。

DAGScheduler内部维护了各类 task / stage / job之间的映射关系表

 

工做流程

 

提交并运行一个Job的基本流程，包括如下步骤

划分Stage

当某个操做触发计算，向DAGScheduler提交做业时，DAGScheduler须要从RDD依赖链最末端的RDD出发，遍历整个RDD依赖链，划分Stage任务阶段，并决定各个Stage之间的依赖关系。Stage的划分是以ShuffleDependency为依据的，也就是说当某个RDD的运算须要将数据进行Shuffle时，这个包含了Shuffle依赖关系的RDD将被用来做为输入信息，构建一个新的Stage，由此为依据划分Stage，能够确保有依赖关系的数据可以按照正确的顺序获得处理和运算。

以GroupByKey操做为例，该操做返回的结果其实是一个ShuffleRDD，当DAGScheduler遍历到这个ShuffleRDD的时候，由于其Dependency是一个ShuffleDependency，因而这个ShuffleRDD的父RDD以及shuffleDependency等对象就被用来构建一个新的Stage，这个Stage的输出结果的分区方式，则由ShuffleDependency中的Partitioner对象来决定。

能够看到，尽管划分和构建Stage的依据是ShuffleDependency，对应的RDD也就是这里的ShuffleRDD，可是这个Stage所处理的数据是从这个shuffleRDD的父RDD开始计算的，只是最终的输出结果的位置信息参考了ShuffleRDD返回的ShuffleDependency里所包含的内容。而shuffleRDD自己的运算操做（其实就是一个获取shuffle结果的过程），是在下一个Stage里进行的。

生成Job，提交Stage

上一个步骤获得一个或多个有依赖关系的Stage，其中直接触发Job的RDD所关联的Stage做为FinalStage生成一个Job实例，这二者的关系进一步存储在resultStageToJob映射表中，用于在该Stage所有完成时作一些后续处理，如报告状态，清理Job相关数据等。

具体提交一个Stage时，首先判断该Stage所依赖的父Stage的结果是否可用，若是全部父Stage的结果均可用，则提交该Stage，若是有任何一个父Stage的结果不可用，则迭代尝试提交父Stage。 全部迭代过程当中因为所依赖Stage的结果不可用而没有提交成功的Stage都被放到waitingStages列表中等待未来被提交

何时waitingStages中的Stage会被从新提交呢，当一个属于中间过程Stage的任务（这种类型的任务所对应的类为ShuffleMapTask）完成之后，DAGScheduler会检查对应的Stage的全部任务是否都完成了，若是是都完成了，则DAGScheduler将从新扫描一次waitingStages中的全部Stage，检查他们是否还有任何依赖的Stage没有完成，若是没有就能够提交该Stage。

此外每当完成一次DAGScheduler的事件循环之后，也会触发一次从等待和失败列表中扫描并提交就绪Stage的调用过程

任务集的提交

每一个Stage的提交，最终是转换成一个TaskSet任务集的提交，DAGScheduler经过TaskScheduler接口提交TaskSet，这个TaskSet最终会触发TaskScheduler构建一个TaskSetManager的实例来管理这个TaskSet的生命周期，对于DAGScheduler来讲提交Stage的工做到此就完成了。而TaskScheduler的具体实现则会在获得计算资源的时候，进一步经过TaskSetManager调度具体的Task到对应的Executor节点上进行运算

任务做业完成状态的监控

要保证相互依赖的job/stage可以获得顺利的调度执行，DAGScheduler就必然须要监控当前Job  / Stage乃至Task的完成状况。这是经过对外（主要是对TaskScheduler）暴露一系列的回调函数来实现的，对于TaskScheduler来讲，这些回调函数主要包括任务的开始结束失败，任务集的失败，DAGScheduler根据这些Task的生命周期信息进一步维护Job和Stage的状态信息。

此外TaskScheduler还能够经过回调函数通知DAGScheduler具体的Executor的生命状态，若是某一个Executor崩溃了，或者因为任何缘由与Driver失去联系了，则对应的Stage的shuffleMapTask的输出结果也将被标志为不可用，这也将致使对应Stage状态的变动，进而影响相关Job的状态，再进一步可能触发对应Stage的从新提交来从新计算获取相关的数据。

任务结果的获取

一个具体的任务在Executor中执行完毕之后，其结果须要以某种形式返回给DAGScheduler，根据任务类型的不一样，任务的结果的返回方式也不一样

对于FinalStage所对应的任务（对应的类为ResultTask）返回给DAGScheduler的是运算结果自己，而对于ShuffleMapTask，返回给DAGScheduler的是一个MapStatus对象，MapStatus对象管理了ShuffleMapTask的运算输出结果在BlockManager里的相关存储信息，而非结果自己，这些存储位置信息将做为下一个Stage的任务的获取输入数据的依据

而根据任务结果的大小的不一样，ResultTask返回的结果又分为两类，若是结果足够小，则直接放在DirectTaskResult对象内，若是超过特定尺寸（默认约10MB）则在Executor端会将DirectTaskResult先序列化，再把序列化的结果做为一个Block存放在BlockManager里，然后将BlockManager返回的BlockID放在IndirectTaskResult对象中返回给TaskScheduler，TaskScheduler进而调用TaskResultGetter将IndirectTaskResult中的BlockID取出并经过BlockManager最终取得对应的DirectTaskResult。固然从DAGScheduler的角度来讲，这些过程对它来讲是透明的，它所得到的都是任务的实际运算结果。

TaskSetManager

前面提到DAGScheduler负责将一组任务提交给TaskScheduler之后，这组任务的调度工做对它来讲就算完成了，接下来这组任务内部的调度逻辑，则是由TaskSetManager来完成的。

TaskSetManager的主要接口包括：

ResourceOffer：根据TaskScheduler所提供的单个Resource资源包括host，executor和locality的要求返回一个合适的Task。TaskSetManager内部会根据上一个任务成功提交的时间，自动调整自身的Locality匹配策略，若是上一次成功提交任务的时间间隔很长，则下降对Locality的要求（例如从最差要求Process  Local下降为最差要求Node Local），反之则提升对Locality的要求。这一动态调整Locality策略基本能够理解为是为了提升任务在最佳Locality的状况下获得运行的机会，由于Resource资源多是在短时间内分批提供给TaskSetManager的，动态调整Locality门槛有助于改善总体的Locality分布状况。

举个例子，若是TaskSetManager内部有a/b两个任务等待调度，a/b两个任务Prefer的节点分别是Host  A 和 Host B， 这时候先有一个Host C的资源以最差匹配为Rack Local的形式提供给TaskSetManager，若是没有内部动态Locality调整机制，那么好比a任务将被调度。接下来在很短的时间间隔内，一个Host  A的资源来到，一样的b任务被调度。 而本来最佳的状况应该是任务b调度给Host C， 而任务a调度给Host A。

固然动态Locality也会带来必定的调度延迟，所以如何设置合适的调整策略也是须要针对实际状况来肯定的。目前能够设置参数包括

spark.locality.wait.process

spark.locality.wait.node

spark.locality.wait.rack

 

即各个Locality级别中TaskSetManager等待分配下一个任务的时间，若是距离上一次成功分配资源的时间间隔超过对应的参数值，则下降匹配要求（即process  -> node -> rack -> any)， 而每当成功分配一个任务时，则重置时间间隔，并更新Locality级别为当前成功分配的任务的Locality级别 

handleSuccessfulTask / handleFailedTask /handleTaskGettingResult ：用于更新任务的运行状态，Taskset Manager在这些函数中除了更新自身维护的任务状态列表等信息，用于剩余的任务的调度之外，也会进一步调用DAGScheduler的函数接口将结果通知给它。

此外，TaskSetManager在调度任务时还可能进一步考虑Speculation的状况，亦即当某个任务的运行时间超过其它任务的运行完成时间的一个特定比例值时，该任务可能被重复调度。目的固然是为了防止某个运行中的Task因为某些特殊缘由（例如所在节点CPU负载太高，IO带宽被占等等）运行特别缓慢拖延了整个Stage的完成时间，Speculation一样须要根据集群和做业的实际状况合理配置，不然可能反而下降集群性能。 

Pool 调度池

前面咱们说了，DAGScheduler负责构建具备依赖关系的任务集，TaskSetManager负责在具体的任务集的内部调度任务，而TaskScheduler负责将资源提供给TaskSetManager供其做为调度任务的依据。可是每一个SparkContext可能同时存在多个可运行的任务集（没有依赖关系），这些任务集之间如何调度，则是由调度池（POOL）对象来决定的，Pool所管理的对象是下一级的Pool或者TaskSetManager对象

TaskSchedulerImpl在初始化过程当中会根据用户设定的SchedulingMode（默认为FIFO）建立一个rootPool根调度池，以后根据具体的调度模式再进一步建立SchedulableBuilder对象，具体的SchedulableBuilder对象的BuildPools方法将在rootPool的基础上完成整个Pool的构建工做。

目前的实现有两种调度模式，对应了两种类型的Pool：

FIFO：先进先出型，FIFO Pool直接管理的是TaskSetManager，每一个TaskSetManager建立时都存储了其对应的StageID，FIFO  pool最终根据StageID的顺序来调度TaskSetManager

FAIR：公平调度，FAIR Pool管理的对象是下一级的POOL，或者TaskSetManager，公平调度的基本原则是根据所管理的Pool/TaskSetManager中正在运行的任务的数量来判断优先级，用户能够设置minShare最小任务数，weight任务权重来调整对应Pool里的任务集的优先程度。当采用公平调度模式时，目前所构建的调度池是两级的结构，即根调度池管理一组子调度池，子调度池进一步管理属于该调度池的TaskSetManager

 公平调度模式的配置经过配置文件来管理，默认使用fairscheduler.xml文件，范例参见conf目录下的模板：

<?xmlversion="1.0"?>
<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>

因为这里的调度池是在SparkContext内部的调度，所以其调度范畴是一个基于该SparkContext的Spark应用程序，正常状况下，多个Spark应用程序之间在调度池层面是没有调度优先级关系的。那么这种调度模式的应用场合是怎样的呢？ 举一个例子就是SparkServer或者SharkServer，做为一个长期运行的SparkContext，他们代理运行了其它连上Server的Spark应用的任务，这样你能够为每一个连接按照用户名指定一个Pool运行，从而实现用户优先级和资源分配的合理调度等。

 

Spark应用之间的调度

 

前面提到调度池只是在SparkContxt内部调度资源，SparkContext之间的调度关系，按照Spark不一样的运行模式，就不必定归Spark所管理的了。

在Mesos和YARN模式下，底层资源调度系统的调度策略由Mesos和YARN所决定，只有在Standalone模式下，Spark  Master按照当前cluster资源是否知足等待列表中的Spark应用 对内存和CPU资源的需求，而决定是否建立一个SparkContext对应的Driver，进而完成Spark应用的启动过程，这能够粗略近似的认为是一种粗颗粒度的有条件的FIFO策略吧

转自：http://blog.csdn.net/colorant/article/details/24010035