【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中本地运行级别从小到达可分为：

进程本地化
节点本地化
无优先位置
机架本地化
任意节点

在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操作，分为两种

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

2. 过程中涉及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中，并将保存的地址序列化后返回，否则直接将运行结果序列化后返回