Driver的任务提交过程
1、Driver程序的代码运行到action操作,触发了SparkContext的runJob方法。
2、SparkContext调用DAGScheduler的runJob函数。
3、DAGScheduler把Job划分stage,然后把stage转化为相应的Tasks,把Tasks交给TaskScheduler。
4、通过TaskScheduler把Tasks添加到任务队列当中,交给SchedulerBackend进行资源分配和任务调度。
5、调度器给Task分配执行Executor,ExecutorBackend负责执行Task。
Spark调度管理
本文主要介绍在单个任务内Spark的调度管理,Spark调度相关概念如下:
- Task(任务):单个分区数据及上的最小处理流程单元。
- TaskSet(任务集):由一组关联的,但互相之间没有Shuffle依赖关系的任务所组成的任务集。
- Stage(调度阶段):一个任务集对应的调度阶段。
- Job(作业):有一个RDD Action生成的一个或多个调度阶段所组成的一次计算作业。
- Application(应用程序):Spark应用程序,由一个或多个作业组成。
各概念间的逻辑关系如下图所示:
Spark的调度管理模块中,最重要的类是DAGScheduler和TaskScheduler,TaskScheduler负责每个具体任务的实际物理调度,DAGScheduler负责将作业拆分成不同阶段的具有依赖关系的多批任务,可以理解为DAGScheduler负责任务的逻辑调度。Spark调度管理示意图如下:
调度阶段的拆分
一个Spark任务提交后,DAGScheduler从RDD依赖链末端的RDD出发,遍历整个RDD依赖链,将Job分解成具有前后依赖关系的多个stage。DAGScheduler是根据ShuffleDependency划分stage的,也就是说当某个RDD的运算需要将数据进行shuffle操作时,这个包含了shuffle依赖关系的RDD将被用来作为输入信息,构建一个新的调度阶段。以此为依据划分调度阶段,可以确保有依赖关系的数据能够按照正确的顺序得到处理和运算。
如何进行Stage划分?下图给出的是对应Spark应用程序代码生成的Stage。它的具体划分依据是根据RDD的依赖关系进行,在遇到宽依赖时将两个RDD划分为不同的Stage。
从上图中可以看到,RDD G与RDD F间的依赖是宽依赖,所以RDD F与 RDD G被划分为不同的Stage,而RDD G 与 RDD 间为窄依赖,因此 RDD B 与 RDD G被划分为同一个Stage。通过这种递归的调用方式,将所有RDD进行划分。
Stage划分算法
由于Spark的算子构建一般都是链式的,这就涉及了要如何进行这些链式计算,Spark的策略是对这些算子,先划分Stage,然后在进行计算。
由于数据是分布式的存储在各个节点上的,所以为了减少网络传输的开销,就必须最大化的追求数据本地性,所谓的数据本地性是指,在计算时,数据本身已经在内存中或者利用已有缓存无需计算的方式获取数据。
Stage划分算法思想
(1)一个Job由多个Stage构成
一个Job可以有一个或者多个Stage,Stage划分的依据就是宽依赖,产生宽依赖的算子:reduceByKey、groupByKey等等
(2)根据依赖关系,从前往后依次执行多个Stage
SparkApplication 中可以因为不同的Action触发众多的Job,也就是说一个Application中可以有很多的Job,每个Job是有一个或者多个Stage构成,后面的Stage依赖前面的Stage,也就是说只有前面的Stage计算完后,后面的Stage才会运行。
(3)Stage的执行时Lazy级别的
所有的Stage会形成一个DAG(有向无环图),由于RDD的Lazy特性,导致Stage也是Lazy级别的,只有遇到了Action才会真正发生作业的执行,在Action之前,Spark框架只是将要进行的计算记录下来,并没有真的执行。
调度阶段的提交
在划分Stage的步骤中会得到一个或多个有依赖关系的Stage,其中直接触发作业的RDD关联的调度阶段被称为FinalStage,DAGScheduler从FinalStage开始生成一个Job。Job和Stage的关系存储在一个映射表中,用于在该调度阶段全部完成时做一些后续处理,如报告状态、清理作业相关数据等。
具体提交一个Stage时,首先判断其依赖的所有父Stage的结果是否可用。如果所有父Stage的结果都可用,则提交该Stage。如果有任何一个父Stage的结果不可用,则尝试迭代提交当前不可用的父Stage。在迭代过程中,父Stage还未运行的Stage都被放到等待队列中,等待将来被提交。
下图是一个具有四个调度阶段的Job的Stage提交顺序:
当一个属于中间过程调度阶段的任务(这种类型的任务所对应的类为ShuffleMapTask)完成后,DAGScheduler会检查对应调度阶段的所有任务是否都完成了。如果完成了,则DAGScheduler将重新扫描一次等待列表中所有的Stage,检查它们是否还有依赖的Stage没有完成。如果所有依赖的Stage都已执行完毕,则提交该Stage。
任务结果的获取
根据任务结果的大小不同,ResultTask返回的结果分为两中形式:
- 如果结果足够小,则直接放在DirectTaskResult对象内。
- 如果超过特定尺寸(默认约10MB),则在Executor端会将DirectTaskResult序列化,将序列化的结果作为一个数据块存放在BlockManager中,然后将BlockManager返回的BlockId放在IndirectTaskResult对象中返回给TaskScheduler,TaskScheduler进而调用TaskResultGetter将IndirectTaskResult中的BlockId取出并通过BlockManager最终取得对应的DirectTaskResult。