Spark调度管理

unitykwdo 2016-10-10

Spark调度管理

本文主要介绍在单个任务内Spark的调度管理,Spark调度相关概念如下:

  • Task(任务):单个分区数据及上的最小处理流程单元。
  • TaskSet(任务集):由一组关联的,但互相之间没有Shuffle依赖关系的任务所组成的任务集。
  • Stage(调度阶段):一个任务集对应的调度阶段。
  • Job(作业):有一个RDD Action生成的一个或多个调度阶段所组成的一次计算作业。
  • Application(应用程序):Spark应用程序,由一个或多个作业组成。

各概念间的逻辑关系如下图所示:

Spark调度管理

Spark的调度管理模块中,最重要的类是DAGScheduler和TaskScheduler,TaskScheduler负责每个具体任务的实际物理调度,DAGScheduler负责将作业拆分成不同阶段的具有依赖关系的多批任务,可以理解为DAGScheduler负责任务的逻辑调度。Spark调度管理示意图如下:
Spark调度管理

调度阶段的拆分

一个Spark任务提交后,DAGScheduler从RDD依赖链末端的RDD出发,遍历整个RDD依赖链,将Job分解成具有前后依赖关系的多个stage。DAGScheduler是根据ShuffleDependency划分stage的,也就是说当某个RDD的运算需要将数据进行shuffle操作时,这个包含了shuffle依赖关系的RDD将被用来作为输入信息,构建一个新的调度阶段。以此为依据划分调度阶段,可以确保有依赖关系的数据能够按照正确的顺序得到处理和运算。

调度阶段的提交

在划分Stage的步骤中会得到一个或多个有依赖关系的Stage,其中直接触发作业的RDD关联的调度阶段被称为FinalStage,DAGScheduler从FinalStage开始生成一个Job。Job和Stage的关系存储在一个映射表中,用于在该调度阶段全部完成时做一些后续处理,如报告状态、清理作业相关数据等。

具体提交一个Stage时,首先判断其依赖的所有父Stage的结果是否可用。如果所有父Stage的结果都可用,则提交该Stage。如果有任何一个父Stage的结果不可用,则尝试迭代提交当前不可用的父Stage。在迭代过程中,父Stage还未运行的Stage都被放到等待队列中,等待将来被提交。

下图是一个具有四个调度阶段的Job的Stage提交顺序:

Spark调度管理

当一个属于中间过程调度阶段的任务(这种类型的任务所对应的类为ShuffleMapTask)完成后,DAGScheduler会检查对应调度阶段的所有任务是否都完成了。如果完成了,则DAGScheduler将重新扫描一次等待列表中所有的Stage,检查它们是否还有依赖的Stage没有完成。如果所有依赖的Stage都已执行完毕,则提交该Stage。

在这里,博主有一个疑问:能否按照DAG划分的Stage的拓扑顺序提交执行Stage?求大家指点迷津。

任务结果的获取

根据任务结果的大小不同,ResultTask返回的结果分为两中形式:

  • 如果结果足够小,则直接放在DirectTaskResult对象内。
  • 如果超过特定尺寸(默认约10MB),则在Executor端会将DirectTaskResult序列化,将序列化的结果作为一个数据块存放在BlockManager中,然后将BlockManager返回的BlockId放在IndirectTaskResult对象中返回给TaskScheduler,TaskScheduler进而调用TaskResultGetter将IndirectTaskResult中的BlockId取出并通过BlockManager最终取得对应的DirectTaskResult。

更多Spark相关教程见以下内容

Spark 的详细介绍:请点这里
Spark 的下载地址:请点这里

相关推荐