【Spark系列4】Task的执行

一、Task的执行流程

1.1、Task执行流程

DAGScheduler将Stage生成TaskSet之后，会将Task交给TaskScheduler进行处理，TaskScheduler负责将Task提交到集群中运行，并负责失败重试，为DAGScheduler返回事件信息等，整体如流程如下：

当任务提交到TaskScheduler时，TaskScheduler会通知SchedulerBackend分配计算资源，SchedulerBackend将所有可用的Executor的资源信息转换成WorkerOffer交给TaskScheduler，WorkerOffer中包含executorId、Executor的hostname、Executor的可用CPU等。TaskScheduler负责根据WorkerOffer在相应的Executor分配TaskSet中的Task，并将Task转换为TaskDescription交给SchedulerBackend。最终有空闲的的CPU的Executor会被分配到一个或者多个TaskDescription，SchedulerBackend将这些TaskDescription提交到对应的Executor中执行。

1.2、集群资源管理

Task运行离不开集群中的计算资源，即在SparkContext初始化过程中创建的Executor资源。在Executor创建完毕后回向SchedulerBackend中注册。Executor在注册时发送的信息包含的内容有：executorId，Executor-Ref引用、Executor的hostname、可用的CPU核数。

SchedulerBackend收到后，会将Executor的注册信息转换为ExecutorData进行保存，并且在SchedulerBackend中使用Map结构保存每个executorId和ExecutorData的关系，ExecutorData中还记录了剩余的可用的CPU核数

在为计算任务分配资源时，只需遍历所有的ExecutorData，分配可用的资源即可。由ExecutorData分配的可用资源使用WorkerOffer表示，WorkerOffer中包含executorId、Executor的hostname、Executor的可用CPU等。

1.3、任务的分配

TaskScheduler在接受到DAGScheduler提交的TaskSet以后，会为每个TaskSet创建一个TaskSetManager，用于管理TaskSet中所有任务的运行。TaskSetManager会根据Task中的最佳运行位置计算TaskSet的所有本地运行级别，本地运行级别决定Task最终在哪个Executor上运行。Spark中本地运行级别从小到达可分为：

1. 进程本地化
2. 节点本地化
3. 无优先位置
4. 机架本地化
5. 任意节点

在TaskSetManager初始化时，根据着5个本地运行级别分别创建5个Map，分别记录其下可以运行的所有Task。这些映射关系的建立，时根据生成Task时Task运行的最佳位置确定的。。在这5种映射关系中，某个Task可能会重复存在于几个本地化级别中。

当有新的TaskSet加入、由Task执行完成、由新的Executor加入时，都会触发SchedulerBackend重新计算可用资源。TaskScheduler根据调度的顺序，依次调度TaskSetManager中的TaskSet，对于每个TaskSet遍历所有本地化级别，从小到大尝试在Executor分配Task，根据每个WorkerOffer的executorId和hostname，使用TaskSetManager判断在当前本地化级别中，是否可以在该Executor或Host上分配任务，直到该本地化级别无法分配Task，再将本地化级别提高一级再次尝分配Task。经过对本地化级别的便利，即可实现WorkerOffer分配任务或将所有待执行的任务分配完成。TaskSet中部分任务分配完成以后会生成一组TaskDescription，每个TaskDescription中包含executorId和Task的其他运行信息。SchedulerBackend根据TaskDescription的executorId，将每个任务封装成LaunchTask消息提交到不同的Executor中

二、Task的执行

Executor收到SchedulerBackend提交的LaunchTask消息后，即可运行该消息中包含的Task。Executor将接收到的Task封装到TaskRunner中，TaskRunner是一个Runnable接口，从而可以将该任务提交到线程池中运行。

2.1、Executor可以并行运行Task的数量

在创建Executor时，每个Executor可能会分配多个CPU核数，而Executor运行的所有任务都是在线程池中运行。Executor运行的时候其本身没有记录CPU使用的情况，对于Executor能够同时运行多少个任务是由SchedulerBackend控制的，SchedulerBackend每在一个Executor中提交一个任务时，便在ExecutorData中减少该Executor可用的CPU核数，直到该Executor生成的WorkerOffer可用的CPU核数为0，便不再为Executor分配任务了。默认每个Task使用一个CPU核心运行，该变量可以通过Spark的配置spark.task.CPUs修改

2.2、Executor中资源共享

当在一个Executor上运行多个Task时，多个Task共享Executor中的SparkEnv的所有组件，共用Executor中分配的内存。如使用Spark广播变量时，每个Executor中会存在一份，Executor所有任务共享这一份变量。当Executor中的BlockManager缓存了某个rdd某分区的数据时，在该Executor上调度使用这个RDD的这个分区的数据的Task执行，可以有效的减少网络加载数据的过程，减少网络传输

2.3、ResultTask运行

在执行ResultTask时，首先会反序列化出该Task执行计算的RDD和对该RDD执行的操作。根据是否涉及Shuffle操作，分为两种

1. 用户编写的RDDtransformation中，不涉及Shuffle操作，一个Job就只涉及一个ResultStage，rdd1直接从数据源中加载

1. 过程中涉及Shuffle操作，划分为两个Stage，rdd1位Shuffle的Reduce阶段。由于DAGScheduler在划分Stage，必先会先计算父Stage，所以执行到ResultStage时，，其父Stage的Map阶段已经完成，并且计算结果已经保存到了BlockManager中，ResultStage中的rdd1之需要根据MapOutputTrackerMaster的计算结果位置信息加载该分区的数据即可

2.4、ShuffleMapTask运行

在计算ShuffleMapTask时，首先会反序列化出Task包含的计算的RDD和划分此Stage的ShulffleDependency。ShulffleDependency包含RDD需要执行分组操作的分区器partitioner,并且通过ShulffleDependency可以获取ShulffleManager的写入器，将本分区的分组计算结果通过写入器写入文件中进行保存。在这个过程中，一个分区的数据生成的多个分组的数据分别属于下游Reduce阶段的不同的分区的数据

ShuffleMapTask中计算的RDD同样为这个Stage中最后的一个RDD。

下图是多个ShuffleMapStage的RDD转换过程

三、Task结果处理

当Executor中Task运行完成时，需要将Task运行结果返回Driver程序，Driver程序根据结果判断该Stage是否计算完成

3.1、ResultTask结果

ResultTask完成后，会将其结果返回直Driver端。根据运行结果的大小返回的结果 被分为直接运行结果和非直接间接运行结果。 当运行结果大于Spark配置的最大直接结果大小的参数时， 会将运行结果保存至当前Executor的BlockManager中，并将保存的地址序列化后返回，否则直接将运行结果序列化后返回

3.2、ShuffleMapTask结果

ShuffleMapTask运行完成后，会将运行结果直接保存至当前Executor的BlockManager中，并将保存结果的位置封装到MapStatus中，最终ShuffleMapTask运行完成结果都为MapStatus类型

3.3、返回至Driver端

Executor将Task的运行结果序列化后，通过Driver的Endpoint-Ref发送至Driver端，Driver的Endpoint收到运行结果后，通知TaskScheduler Task运行完成