8.4　Driver如何管理ShuffleMapTask和ResultTask的处理结果

Spark Job中，根据Task所处Stage的位置，我们将Task分为两类：第一类叫shuffleMapTask，指Task所处的Stage不是最后一个Stage，也就是Stage的计算结果还没有输出，而是通过Shuffle交给下一个Stage使用；第二类叫resultTask，指Task所处Stage是DAG中最后的一个Stage，也就是Stage计算结果需要进行输出等操作，计算到此为止已经结束。简单地说，Spark Job中除了最后一个Stage的Task叫resultTask，其他所有Task都叫ShuffleMapTask。

8.4.1　ShuffleMapTask执行结果和Driver的交互原理及源码详解

Driver中的CoarseGrainedSchedulerBackend给CoarseGrainedExecutorBackend发送launchTasks消息，CoarseGrainedExecutorBackend收到launchTasks消息以后会调用executor.launchTask。通过launchTask执行Task，launchTask方法中根据传入的参数：taskId、尝试次数、任务名称、序列化后的任务创建一个TaskRunner，在threadPool中执行TaskRunner。TaskRunner内部会先做一些准备工作，如反序列化Task的依赖，通过网络获取需要的文件、Jar等；然后调用反序列化后的Task.run方法来执行任务并获得执行结果。

其中，Task的run方法调用的时候会导致Task的抽象方法runTask的调用，Task.scala的runTask方法是一个抽象方法。Task包括ResultTask、ShuffleMapTask两种Task，抽象runTask方法具体的实现由子类的runTask实现。ShuffleMapTask的runTask实际运行的时候会调用RDD的iterator，然后针对Partition进行计算。

ShuffleMapTask.scala的源码如下：

ShuffleMapTask方法中调用ShuffleManager写入器writer方法，在write时最终计算会调用RDD的compute方法。通过writer.stop(success = true).get，如果写入成功，就返回MapStatus结果值。

SortShuffleWriter.scala的源码如下：

回到TaskRunner的run方法，把task.run执行结果通过resultSer.serialize(value)序列化，生成一个directResult。然后根据大小判断不同的结果赋值给serializedResult，传回给Driver。

（1）如果任务执行结果特别大，超过1GB，日志就提示超出任务大小限制，返回元数据ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))。

Executor.scala的源码如下：

（2）如果任务执行结果小于1GB，大于maxDirectResultSize（128MB），就放入blockManager，返回元数据ser.serialize(new IndirectTaskResult[Any](blockId, resultSize))。

Executor.scala的源码如下：

（3）如果任务执行结果小于128MB，就直接返回serializedDirectResult。

Executor.scala的源码如下：

接下来，TaskRunner的run方法中调用execBackend.statusUpdate(taskId，TaskState.FINISHED，serializedResult)给Driver发送一个消息，消息中将taskId、TaskState.FINISHED、serializedResult传进去。这里，execBackend是CoarseGrainedExecutorBackend。

Executor.scala的源码如下：

CoarseGrainedExecutorBackend的statusUpdate方法的源码如下：

CoarseGrainedExecutorBackend给DriverEndpoint发送StatusUpdate来传输执行结果。DriverEndpoint是一个ThreadSafeRpcEndpoint消息循环体，模式匹配收到StatusUpdate消息，调用scheduler.statusUpdate(taskId, state, data.value)方法执行。这里的scheduler是TaskSchedulerImpl。

CoarseGrainedSchedulerBackend.scala的DriverEndpoint的源码如下：

DriverEndpoint会把执行结果传递给TaskSchedulerImpl处理，交给TaskResultGetter内部，通过线程去分别处理Task执行成功和失败时的不同情况，然后告诉DAGScheduler任务处理结束的状况。

TaskSchedulerImpl.scala的statusUpdate的源码如下：

TaskResultGetter.scala的enqueueSuccessfulTask方法中，开辟一条新线程处理成功任务，对结果进行相应的处理后调用scheduler.handleSuccessfulTask。

TaskSchedulerImpl的handleSuccessfulTask的源码如下：

TaskSchedulerImpl的handleSuccessfulTask交给TaskSetManager调用handleSuccessfulTask。

TaskSetManager的handleSuccessfulTask的源码如下：

handleSuccessfulTask方法中调用sched.dagScheduler.taskEnded，taskEnded由TaskSetManager调用，汇报任务完成或者失败。将任务完成的事件CompletionEvent放入eventProcessLoop事件处理循环中。

DAGScheduler.scala的源码如下：

由事件循环线程读取消息，并调用DAGSchedulerEventProcessLoop.onReceive方法进行消息处理。

DAGScheduler.scala的源码如下：

onReceive中调用doOnReceive(event)方法，模式匹配到CompletionEvent，调用dagScheduler.handleTaskCompletion方法。

DAGScheduler.scala的源码如下：

DAGScheduler.handleTaskCompletion中task执行成功的情况，根据ShuffleMapTask和ResultTask两种情况分别处理。其中，ShuffleMapTask将MapStatus汇报给MapOutTracker。

Spark 2.2.1版本的DAGScheduler的handleTaskCompletion的源码如下：

Spark 2.4.3版本的DAGScheduler的handleTaskCompletion的源码与Spark 2.2.1版本相比具有如下特点。

　上段代码中第11行之后新增代码，从shuffleStage.pendingPartitions中去掉分区ID。

　上段代码中删掉第15～21行。

　上段代码中删掉第25～32行，新增mapOutputTracker.registerMapOutput进行注册的代码。

　上段代码中删掉第46～49行，新增mapOutputTracker.incrementEpoch的代码。

　上段代码中删掉第61～67行，将其封装为一个方法，新增markMapStageJobsAs-Finished(shuffleStage)的代码。

8.4.2　ResultTask执行结果与Driver的交互原理及源码详解

Task的run方法调用的时候会导致Task的抽象方法runTask的调用，Task.scala的runTask方法是一个抽象方法。Task包括ResultTask、ShuffleMapTask两种Task，抽象runTask方法具体的实现由子类的runTask实现。ResultTask的runTask具体实现的源码如下。

ResultTask.scala的runTask的源码如下：

而ResultTask的runTask方法中反序列化生成func函数，最后通过func函数计算出最终的结果。

ResultTask执行结果与Driver的交互过程同ShuffleMapTask类似，最终，DAGScheduler.handleTaskCompletion中Task执行结果，根据ShuffleMapTask和ResultTask两种情况分别处理。其中，ResultTask的处理结果如下所示。

DAGScheduler的handleTaskCompletion的源码如下：

Driver端的DAGScheduler的MapOutputTracker把shuffleMapTask执行的结果交给ResultTask，ResultTask根据前面Stage的执行结果进行shuffle后产生整个Job最后的结果。