spark调度管理原理

1、Spark调度管理原理 Spark调度管理原理 Spark程序得以运转的核心是它的调度管理逻辑，在Spark底层物理资源调度到上层应用。处于核心位置的作业调度管理模块是整个程序运作的枢纽。4.1 Spark作业调度管理概述 在Spark作业调度系统中，调度的前提是判断多个作业任务的依赖关系，这些作业任务之间可能存在因果的依赖关系。适合用DAG有向无环图来表示。 在作业调度的相关类中，最重要的就是DAGScheduler（基于DAG图的调度类）。 TaskScheduler负责每个具体任务的实际物理调度，DAGScheduler负责将作业拆分成不同阶段的具有依赖关系的多批任务，可以理解为DAGS

2、cheduler负责任务的逻辑调度。4.2 Spark调度的相关基本概念Task（任务）：单个分区数据集上的最小处理流程单元。TaskSet （任务集）：由一组关联的，但相互之间没有Shuffle依赖关系的任务所组成的任务集。Stage（调度阶段）：一个任务集对应的调度阶段。Job（作业）：由一个或多个调度阶段所组成的一次计算作业。Application（应用程序）：Spark应用程序，由一个或多个作业组成。 4.3 作业调度模块顶层逻辑概述 在Spark应用程序中，用户代码基本是基于RDD的一系列计算操作。在实际运行时，这些计算操作是延迟执行的，并不是所有的RDD操作都会触发Spark向集群

3、提交实际作业，只有一些需要返回数据或者向外部输出数据的操作才会触发实际的计算工作，其他的变换操作只是生成对应的RDD关系链，用来记录依赖关系和所需执行的运算。 DAGScheduler最重要的任务之一就是计算作业和任务的依赖关系，制定调度逻辑。DAGScheduler中作业调度的发起，对外暴露的两个主要的入口其中之一是submitJob，另一个是runJob。两者的区别在于前者返回一个Jobwaiter对象，可以用在异步调用中，用来判断作业完成或者取消作业，而后者则在内部调用submitJob，阻塞等待直到作业完成（或失败）。4.3 作业调度模块顶层逻辑概述 DAGScheduler在Spar

4、kContext初始化的过程中被实例化，一个SparkContext对应创建一个DAGScheduler。 DAGScheduler在初始化过程中会创建一个DAGSchedulerEventProcessActor实例来处理各种DAGSchedulerEvent事件，这些事件包括作业的提交、任务状态的变化、监控等，如程序4-2所示，DAGSchedulerEventProcessActor的Receive函数是DAGScheduler中处理这些事件的入口，可以看到各种待处理的事件。4.3 作业调度模块顶层逻辑概述4.3 作业调度模块顶层逻辑概述 eventProcessActor是DAGSch

5、eduler的私有成员，通过DAGScheduler封装好的公共函数接口间接地向eventProcessActor发送相关消息，如在前面提到的submitJob，如程序4-3所示。4.3 作业调度模块顶层逻辑概述 DAGScheduler内部维护了各种“任务/调度阶段/作业”的状态和相互之间的映射关系表，用于在任务状态更新、集群状态更新等各种情况下，正确的维护作业的运行逻辑。4.4 作业调度具体工作流程 每个作业从提交到完成，都要经历多个步骤，拆分成以任务为最小单位，按照一定的逻辑依赖关系依次提交执行，并返回结果。4.4 作业调度具体工作流程4.4.1 调度阶段的拆分4.4.1 调度阶段的拆分

6、4.4.1 调度阶段的拆分 以GroupByKey操作为例，该操作返回的结果实际上是一个ShuffleRDD，当DAGScheduler遍历到这个ShuffleRDD的时候，因为其依赖关系是一个ShuffleDependency，于是这个ShuffleRDD的父RDD以及ShuffleDependency等对象就被用来构建一个新的Stage，这个Stage的输出结果的分区方式，则由ShuffleDependency 中的Partitioner对象决定。 尽管划分和构建Stage的依据是ShuffleDependency，在前面的例子中对应的RDD是ShuffleRDD，但是这个Stage所处理

7、的数据是从这个ShuffleRDD的父RDD开始的，只是最终的输出结果位置信息参考了ShuffleRDD返回的ShuffleDependency里所包含的内容。而ShuffleRDD本身的运算操作，是在下一个Stage里进行的。下面以pagerank的程序4-5为例说明。4.4.1 调度阶段的拆分4.4.1 调度阶段的拆分4.4.2 调度阶段的提交4.4.2 调度阶段的提交 将重新扫描一次等待列表中的所有调度阶段的列表，检查它们是否还有任何依赖的调度阶段没有完成，如果没有，说明该调度阶段处于就绪状态，就可以再次尝试提交该阶段了。4.4.3 任务集的提交调度阶段的提交，最终会被转换成一个任务集的

8、提交，DAGScheduler通过TaskScheduler接口提交任务集，这个任务集最终会触发TaskScheduler构建一个TaskSetManager的实例来管理这个任务集的生命周期，对于DAGScheduler来说，提交调度阶段的工作到此就完成了。而TaskScheduler的具体实现则会在得到计算资源的时候，进一步通过TaskSetManager调度具体的任务到对应的Executor节点上进行运算。4.4.4 完成状态的监控 要保证相互依赖的作业调度阶段能够得到顺利的调度执行，DAGScheduler需要监控当前作业调度阶段乃至任务的完成情况。这通过对外暴露一系列的回调函数来实现的

9、，对于TaskScheduler来说，这些回调函数主要包括任务的开始结束失败、任务集的失败，DAGScheduler根据这些任务的生命周期信息进一步维护作业和调度阶段的状态信息。 TaskScheduler还可以通过回调函数通知DAGScheduler具体的Executor的生命状态，如果某一个Executor崩溃了，则对应的调度阶段任务集的ShuffleMapTask的输出结果也将标志为不可用，这将导致对应任务集状态的变更，进而重新执行相关计算任务，以获取丢失的相关数据。4.4.5 任务结果的获取一个具体的任务在Executor中执行完毕后，其结果需要以某种形式返回给DAGScheduler

10、，根据任务类型的不同，任务结果的返回方式也不同。对于FinalStage所对应的任务，返回给DAGScheduler的是运算结果本身，而对于中间调度阶段对应的任务ShuffleMapTask，返回给DAGScheduler的是一个MapStatus里的相关存储信息，而非结果本身，这些存储位置信息将作为下一个调度阶段的任务获取输入数据的依据。根据任务结果大小的不同，ResultTask返回的结果又分为两类，如果结果足够小，则直接放在DirectTaskResult对象内中，如果超过特定尺寸则在Executor端会将DirectTaskResult先序列化，再把序列化的结果作为一个数据块存放在Bl

11、ockManager中，然后将BlockManager返回的BlockID放在IndirectTaskResult对象中返回给TaskScheduler，TaskScheduler进而调用TaskResultGetter将IndirectTaskResult中的BlockID取出并通过BlockManager最终取得对应的DirectTaskResult。4.5 任务集管理模块TaskSetManager的主要接口：ResourceOffer：根据TaskScheduler所提供的单个Resource资源以及任务的host、executor和locality本地性的要求返回一个合适的任务。handleSuccessfulTask/handleFailedTask/handleTaskGettingResult：用于更新任务的运行状态，TaskSetManager在这些函数中除了更新自身维护的任务状态列表等信息以用于剩余的任务调度外，还会进一步调用DAGScheduler的函数接口将结果通知给它。4.6 调度池和调度模式分析每个SparkContext可能又同时存在多个可运行的任务集，这些任务集之间如何调度，则是