8.4 ShuffleMapTask和ResultTask处理结果是如何被Driver管理的
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的源码如下。
1. override def runTask(context: TaskContext): MapStatus = {
2. ......
3. val ser = SparkEnv.get.closureSerializer.newInstance()
4. val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _,
_])](
5. ......
6. val manager = SparkEnv.get.shuffleManager
7. writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId,
context)
8. writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator
[_ <: Product2[Any, Any]]])
9. writer.stop(success = true).get
10. ......
ShuffleMapTask方法中调用ShuffleManager写入器writer方法，在write时最终计算会调用RDD的compute方法。通过writer.stop(success = true).get，如果写入成功，就返回MapStatus结果值。
SortShuffleWriter.scala的源码如下。
1. override def write(records: Iterator[Product2[K, V]]): Unit = {
2. ......
3. val blockId = ShuffleBlockId(dep.shuffleId, mapId,
IndexShuffleBlockResolver.NOOP_REDUCE_ID)
4. val partitionLengths = sorter.writePartitionedFile(blockId, tmp)
5. shuffleBlockResolver.writeIndexFileAndCommit(dep.shuffleId, mapId,
partitionLengths, tmp)
6. mapStatus = MapStatus(blockManager.shuffleServerId, partitionLengths)
7. ......
8. override def stop(success: Boolean): Option[MapStatus] = {
9. ......
10. if (success) {
11. return Option(mapStatus)
12. } else {
13. return None
14. }
15. ......
回到TaskRunner的run方法，把task.run执行结果通过resultSer.serialize(value)序列化，生成一个directResult。然后根据大小判断不同的结果赋值给serializedResult，传回给Driver。
（1）如果任务执行结果特别大，超过1GB，日志就提示超出任务大小限制，返回元数据ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))。
Executor.scala的源码如下。
1. if (maxResultSize > 0 && resultSize > maxResultSize) {
2. logWarning(s"Finished $taskName (TID $taskId). Result is larger
than maxResultSize " + s"(${Utils.bytesToString(resultSize)} >
${Utils.bytesToString (maxResultSize)}), " + s"dropping it.")
3.
4. ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId
(taskId), resultSize))
（2）如果任务执行结果小于1GB，大于maxDirectResultSize（128MB），就放入blockManager，返回元数据ser.serialize(new IndirectTaskResult[Any](blockId, resultSize))。
Executor.scala的源码如下。
1. .......
2. } else if (resultSize > maxDirectResultSize) {
3. val blockId = TaskResultBlockId(taskId)
4. env.blockManager.putBytes(
5. blockId,
6. new ChunkedByteBuffer(serializedDirectResult.duplicate()),
7. StorageLevel.MEMORY_AND_DISK_SER)
8. logInfo(
9. s"Finished $taskName (TID $taskId). $resultSize bytes result
sent via BlockManager)")
10. ser.serialize(new IndirectTaskResult[Any](blockId, resultSize))
（3）如果任务执行结果小于128MB，就直接返回serializedDirectResult。
Executor.scala的源码如下。
1. .......
2. } else {
3. logInfo(s"Finished $taskName (TID $taskId). $resultSize bytes
result sent to driver")
4. serializedDirectResult
5. ......
接下来，TaskRunner的run方法中调用execBackend.statusUpdate(taskId，TaskState.FINISHED，serializedResult)给Driver发送一个消息，消息中将taskId、TaskState.FINISHED、serializedResult传进去。这里，execBackend是CoarseGrainedExecutorBackend。
Executor.scala的源码如下。
1. override def run(): Unit = {
2. .....
3. execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult)
4. ......
CoarseGrainedExecutorBackend的statusUpdate方法的源码如下。
1. override def statusUpdate(taskId: Long, state: TaskState, data:
ByteBuffer) {
2. val msg = StatusUpdate(executorId, taskId, state, data)
3. driver match {
4. case Some(driverRef) => driverRef.send(msg)
5. case None => logWarning(s"Drop $msg because has not yet connected
to driver")
6. }
7. }
CoarseGrainedExecutorBackend给DriverEndpoint发送StatusUpdate来传输执行结果。DriverEndpoint是一个ThreadSafeRpcEndpoint消息循环体，模式匹配收到StatusUpdate消息，调用scheduler.statusUpdate(taskId, state, data.value)方法执行。这里的scheduler是TaskSchedulerImpl。
CoarseGrainedSchedulerBackend.scala的DriverEndpoint的源码如下。
1. override def receive: PartialFunction[Any, Unit] = {
2. case StatusUpdate(executorId, taskId, state, data) =>
3. scheduler.statusUpdate(taskId, state, data.value)
DriverEndpoint会把执行结果传递给TaskSchedulerImpl处理，交给TaskResultGetter内部，通过线程去分别处理Task执行成功和失败时的不同情况，然后告诉DAGScheduler任务处理结束的状况。
TaskSchedulerImpl.scala的statusUpdate的源码如下。
1. def statusUpdate(tid: Long, state: TaskState, serializedData: ByteBuffer) {
2. ........
3. if (TaskState.isFinished(state)) {
4. cleanupTaskState(tid)
5. taskSet.removeRunningTask(tid)
6. if (state == TaskState.FINISHED) {
7. taskResultGetter.enqueueSuccessfulTask(taskSet, tid,
serializedData)
8. } else if (Set(TaskState.FAILED, TaskState.KILLED,
TaskState.LOST).contains(state)) {
9. taskResultGetter.enqueueFailedTask(taskSet, tid, state,
serializedData)
10. }
11. }
TaskResultGetter.scala的enqueueSuccessfulTask方法中，开辟一条新线程处理成功任务，对结果进行相应的处理后调用scheduler.handleSuccessfulTask。
TaskSchedulerImpl的handleSuccessfulTask的源码如下。
1. def handleSuccessfulTask(
2. taskSetManager: TaskSetManager,
3. tid: Long,
4. taskResult: DirectTaskResult[_]): Unit = synchronized {
5. taskSetManager.handleSuccessfulTask(tid, taskResult)
6. }
TaskSchedulerImpl的handleSuccessfulTask交给TaskSetManager调用handleSuccessfulTask。
TaskSetManager的handleSuccessfulTask的源码如下。
1. def handleSuccessfulTask(tid: Long, result: DirectTaskResult[_]): Unit = {
2. ......
3. sched.dagScheduler.taskEnded(tasks(index), Success, result.value(),
result.accumUpdates, info)
4. ......
5.
handleSuccessfulTask方法中调用sched.dagScheduler.taskEnded，taskEnded由TaskSetManager调用，汇报任务完成或者失败。将任务完成的事件CompletionEvent放入eventProcessLoop事件处理循环中。
DAGScheduler.scala的源码如下。
1. def taskEnded(
2. task: Task[_],
3. reason: TaskEndReason,
4. result: Any,
5. accumUpdates: Seq[AccumulatorV2[_, _]],
6. taskInfo: TaskInfo): Unit = {
7. eventProcessLoop.post(
8. CompletionEvent(task, reason, result, accumUpdates, taskInfo))
9. }
由事件循环线程读取消息，并调用DAGSchedulerEventProcessLoop.onReceive方法进行消息处理。
DAGScheduler.scala的源码如下。
1. override def onReceive(event: DAGSchedulerEvent): Unit = {
2. val timerContext = timer.time()
3. try {
4. doOnReceive(event)
5. } finally {
6. timerContext.stop()
7. }
8. }
onReceive中调用doOnReceive(event)方法，模式匹配到CompletionEvent，调用dagScheduler.handleTaskCompletion方法。
DAGScheduler.scala的源码如下。
1. private def doOnReceive(event: DAGSchedulerEvent): Unit = event match {
2. case JobSubmitted(jobId, rdd, func, partitions, callSite, listener,
properties) =>
3. dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions,
callSite, listener, properties)
4. ......
5. case completion: CompletionEvent =>
6. dagScheduler.handleTaskCompletion(completion)
7. .....
DAGScheduler.handleTaskCompletion中task执行成功的情况，根据ShuffleMapTask和ResultTask两种情况分别处理。其中，ShuffleMapTask将MapStatus汇报给MapOutTracker。
Spark 2.1.1版本的DAGScheduler的handleTaskCompletion的源码如下。
1. private[scheduler] def handleTaskCompletion(event: CompletionEvent) {
2. ......
3. val stage = stageIdToStage(task.stageId)
4. event.reason match {
5. case Success =>
6. stage.pendingPartitions -= task.partitionId
7. task match {
8. ......
9. case smt: ShuffleMapTask =>
10. val shuffleStage = stage.asInstanceOf[ShuffleMapStage]
11. updateAccumulators(event)
12. val status = event.result.asInstanceOf[MapStatus]
13. val execId = status.location.executorId
14. logDebug("ShuffleMapTask finished on " + execId)
15. if (failedEpoch.contains(execId) && smt.epoch <= failedEpoch
(execId)) {
16. logInfo(s"Ignoring possibly bogus $smt completion from
executor $execId")
17. } else {
18. shuffleStage.addOutputLoc(smt.partitionId, status)
19. }
20.
21. if (runningStages.contains(shuffleStage) && shuffleStage.
pendingPartitions.isEmpty) {
22. markStageAsFinished(shuffleStage)
23. logInfo("looking for newly runnable stages")
24. logInfo("running: " + runningStages)
25. logInfo("waiting: " + waitingStages)
26. logInfo("failed: " + failedStages)
27. /**
* 我们设置为true，来递增纪元编号，以防止map输出重新计算。在这种情
* 况下，一些节点可能已经缓存了损坏的位置（从我们检测到的错误），将需
* 要纪元编号递增来重取它们。待办事项：如果这不是第一次，那么只增加纪
* 元编号，我们注册了map输出
28. */
29. mapOutputTracker.registerMapOutputs(
30. shuffleStage.shuffleDep.shuffleId,
31. shuffleStage.outputLocInMapOutputTrackerFormat(),
32. changeEpoch = true)
33. clearCacheLocs()
34. if (!shuffleStage.isAvailable) {
35. //有些任务已经失败了，重新提交shuffleStage
36. //待办事项：低级调度器也应该处理这个问题
37. logInfo("Resubmitting " + shuffleStage + " (" +
shuffleStage.name +
38. ") because some of its tasks had failed: " +
39. shuffleStage.findMissingPartitions().mkString(", "))
40. submitStage(shuffleStage)
41. } else {
42. //标识任何map阶段的作业都在这个阶段等待完成
43. if (shuffleStage.mapStageJobs.nonEmpty) {
44. val stats = mapOutputTracker.getStatistics(shuffleStage.
shuffleDep)
45. for (job <- shuffleStage.mapStageJobs) {
46. markMapStageJobAsFinished(job, stats)
47. }
48. }
49. submitWaitingChildStages(shuffleStage)
50. }
51. }
52. }
Spark 2.2.0版本的DAGScheduler的handleTaskCompletion的源码与Spark 2.1.1版本相比具有如下特点。
上段代码中第6行删掉stage.pendingPartitions -= task.partitionId。
上段代码中第14行之后新增if的逻辑判断，如果stageIdToStage(task.stageId). latestInfo.attemptId等于task.stageAttemptId，则执行shuffleStage.pendingPartitions -= task.partitionId。
1. case smt: ShuffleMapTask =>
2. ......
3. if (stageIdToStage(task.stageId).latestInfo.attemptId ==
task.stageAttemptId) {
4. /**
* 任务Task是当前stage正在尝试做的。因为任务在TaskSetManager下顺利完成，其
* 标记为不再等待TaskSetManager把任务完成，当任务的epoch较小时，需忽略输出。
* 在这种情况下，挂起的分区为空时，仍然会丢失输出位置，这将导致DAGScheduler 重新
* 提交下面的stage
5. */
6. shuffleStage.pendingPartitions -= task.partitionId
7. }
8. ......
8.4.2 ResultTask执行结果与Driver的交互原理及源码详解
Task的run方法调用的时候会导致Task的抽象方法runTask的调用，Task.scala的runTask方法是一个抽象方法。Task包括ResultTask、ShuffleMapTask两种Task，抽象runTask方法具体的实现由子类的runTask实现。ResultTask的runTask具体实现的源码如下。
ResultTask.scala的runTask的源码如下。
1. override def runTask(context: TaskContext): U = {
2. ......
3. //反序列RDD和func处理函数
4. val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T])
=> U)](
5. ......
6. func(context, rdd.iterator(partition, context))
7. }
而ResultTask的runTask方法中反序列化生成func函数，最后通过func函数计算出最终的结果。
ResultTask执行结果与Driver的交互过程同ShuffleMapTask类似，最终，DAGScheduler.handleTaskCompletion中Task执行结果，根据ShuffleMapTask和ResultTask两种情况分别处理。其中，ResultTask的处理结果如下。
DAGScheduler的handleTaskCompletion的源码如下。
1. case rt: ResultTask[_, _] =>
2. //因为是ResultTask的一部分，所以对应为ResultStage
3. //待办事宜：这一功能进行重构，接受ResultStage
4. val resultStage = stage.asInstanceOf[ResultStage]
5. resultStage.activeJob match {
6. case Some(job) =>
7. if (!job.finished(rt.outputId)) {
8. updateAccumulators(event)
9. job.finished(rt.outputId) = true
10. job.numFinished += 1
11. //如果整个作业完成，就删除
12. if (job.numFinished == job.numPartitions) {
13. markStageAsFinished(resultStage)
14. cleanupStateForJobAndIndependentStages(job)
15. listenerBus.post(
16. SparkListenerJobEnd(job.jobId, clock.getTimeMillis(),
JobSucceeded))
17. }
18.
19. //taskSucceeded 运行用户代码可能会抛出一个异常
20. try {
21. job.listener.taskSucceeded(rt.outputId, event.result)
22. } catch {
23. case e: Exception =>
24. //待办事项：可能我们要标记resultStage 失败？
25. job.listener.jobFailed(new SparkDriverExecution-
Exception(e))
26. }
27. }
28. case None =>
29. logInfo("Ignoring result from " + rt + " because its job has
finished")
30. }
Driver端的DAGScheduler的MapOutputTracker把shuffleMapTask执行的结果交给ResultTask，ResultTask根据前面Stage的执行结果进行shuffle后产生整个Job最后的结果。