Spark 源码系列（三）作业运行过程

上一章讲了 RDD 的转换，但是没讲作业的运行，它和 Driver Program 的关系是啥，和 RDD 的关系是啥？

官方给的例子里面，一执行 collect 方法就能出结果，那我们就从 collect 开始看吧，进入 RDD，找到 collect 方法。

def collect(): Array[T] = {
    val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
    Array.concat(results: _*)
  }
复制代码

它进行了两个操作：

1、调用 SparkContext 的 runJob 方法，把自身的引用传入去，再传了一个匿名函数（把 Iterator 转换成 Array 数组）

2、把 result 结果合并成一个 Array，注意 results 是一个 Array[Array[T]] 类型，所以第二句的那个写法才会那么奇怪。这个操作是很重的一个操作，如果结果很大的话，这个操作是会报 OOM 的，因为它是把结果保存在 Driver 程序的内存当中的 result 数组里面。

我们点进去 runJob 这个方法吧。

val callSite = getCallSite
val cleanedFunc = clean(func)
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, allowLocal, resultHandler, localProperties.get)
rdd.doCheckpoint()
复制代码

追踪下去，我们会发现经过多个不同的 runJob 同名函数调用之后，执行 job 作业靠的是 dagScheduler，最后把结果通过 resultHandler 保存返回。

DAGScheduler 如何划分作业

好的，我们继续看 DAGScheduler 的 runJob 方法，提交作业，然后等待结果，成功什么都不做，失败抛出错误，我们接着看 submitJob 方法。

val jobId = nextJobId.getAndIncrement()
    val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
    // 记录作业成功与失败的数据结构，一个作业的Task数量是和分片的数量一致的，Task成功之后调用resultHandler保存结果。
    val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
eventProcessActor ! JobSubmitted(jobId, rdd, func2, partitions.toArray, allowLocal, callSite, waiter, properties)
复制代码

走到这里，感觉有点儿绕了，为什么到了这里，还不直接运行呢，还要给 eventProcessActor 发送一个 JobSubmitted 请求呢，new 一个线程和这个区别有多大？

不管了，搜索一下 eventProcessActor 吧，结果发现它是一个 DAGSchedulerEventProcessActor，它的定义也在 DAGScheduler 这个类里面。它的 receive 方法里面定义了 12 种事件的处理方法，这里我们只需要看

JobSubmitted 的就行，它也是调用了自身的 handleJobSubmitted 方法。但是这里很奇怪，没办法打断点调试，但是它的结果倒是能返回的，因此我们得用另外一种方式，打开 test 工程，找到 scheduler 目录下的 DAGSchedulerSuite 这个类，我们自己写一个 test 方法，首先我们要在 import 那里加上 import org.apache.spark.SparkContext._ ，然后加上这一段测试代码。

test("run shuffle") {
    val rdd1 = sc.parallelize(1 to 100, 4)
    val rdd2 = rdd1.filter(_ % 2 == 0).map(_ + 1)
    val rdd3 = rdd2.map(_ - 1).filter(_ < 50).map(i => (i, i))
    val rdd4 = rdd3.reduceByKey(_ + _)
    submit(rdd4, Array(0,1,2,3))
    complete(taskSets(0), Seq(
      (Success, makeMapStatus("hostA", 1)),
      (Success, makeMapStatus("hostB", 1))))
    complete(taskSets(1), Seq((Success, 42)))
    complete(taskSets(2), Seq(
      (Success, makeMapStatus("hostA", 2)),
      (Success, makeMapStatus("hostB", 2))))
    complete(taskSets(3), Seq((Success, 68)))
  }
复制代码

这个例子的重点还是 shuffle 那块，另外也包括了 map 的多个转换，大家可以按照这个例子去测试下。

我们接着看 handleJobSubmitted 吧。

var finalStage: Stage = null
try {
    finalStage = newStage(finalRDD, partitions.size, None, jobId, Some(callSite))
} catch {
    // 错误处理，告诉监听器作业失败，返回....
}
if (finalStage != null) {
    val job = new ActiveJob(jobId, finalStage, func, partitions, callSite, listener, properties)
    clearCacheLocs()
    if (allowLocal && finalStage.parents.size == 0 && partitions.length == 1) {
        // 很短、没有父stage的本地操作，比如 first() or take() 的操作本地执行.
        listenerBus.post(SparkListenerJobStart(job.jobId, Array[Int](), properties))
        runLocally(job)
    } else {
        // collect等操作走的是这个过程，更新相关的关系映射，用监听器监听，然后提交作业
        jobIdToActiveJob(jobId) = job
        activeJobs += job
        resultStageToJob(finalStage) = job
        listenerBus.post(SparkListenerJobStart(job.jobId, jobIdToStageIds(jobId).toArray, properties))
        // 提交stage
        submitStage(finalStage)
    }
}
// 提交stage
submitWaitingStages()
复制代码

从上面这个方法来看，我们应该重点关注 newStage 方法、submitStage 方法和 submitWaitingStages 方法。

我们先看 newStage，它得到的结果叫做 finalStage，挺奇怪的哈，为啥？先看吧

val id = nextStageId.getAndIncrement()
    val stage = new Stage(id, rdd, numTasks, shuffleDep, getParentStages(rdd, jobId), jobId, callSite)
    stageIdToStage(id) = stage
    updateJobIdStageIdMaps(jobId, stage)
    stageToInfos(stage) = StageInfo.fromStage(stage)
    stage
复制代码

可以看出来 Stage 也没有太多的东西可言，它就是把 rdd 给传了进去，tasks 的数量，shuffleDep 是空，parentStage。

那它的 parentStage 是啥呢？

private def getParentStages(rdd: RDD[_], jobId: Int): List[Stage] = {
    val parents = new HashSet[Stage]
    val visited = new HashSet[RDD[_]]
    def visit(r: RDD[_]) {
      if (!visited(r)) {
        visited += r
        // 在visit函数里面，只有存在ShuffleDependency的，parent才通过getShuffleMapStage计算出来
        for (dep <- r.dependencies) {
          dep match {
            case shufDep: ShuffleDependency[_,_] =>
              parents += getShuffleMapStage(shufDep, jobId)
            case _ =>
              visit(dep.rdd)
          }
        }
      }
    }
    visit(rdd)
    parents.toList
  }
复制代码

它是通过不停的遍历它之前的 rdd，如果碰到有依赖是 ShuffleDependency 类型的，就通过 getShuffleMapStage 方法计算出来它的 Stage 来。

那我们就开始看 submitStage 方法吧。

private def submitStage(stage: Stage) {
        //...
        val missing = getMissingParentStages(stage).sortBy(_.id)
        logDebug("missing: " + missing)
        if (missing == Nil) {
          // 没有父stage，执行这stage的tasks
          submitMissingTasks(stage, jobId.get)
          runningStages += stage
        } else {
         // 提交父stage的task，这里是个递归，真正的提交在上面的注释的地方
          for (parent <- missing) {
            submitStage(parent)
          }
          // 暂时不能提交的stage，先添加到等待队列
          waitingStages += stage
        }
      }
  }
复制代码

这个提交 stage 的过程是一个递归的过程，它是先要把父 stage 先提交，然后把自己添加到等待队列中，直到没有父 stage 之后，就提交该 stage 中的任务。等待队列在最后的 submitWaitingStages 方法中提交。

这里我引用一下上一章当中我所画的那个图来表示这个过程哈。

从 getParentStages 方法可以看出来，RDD 当中存在 ShuffleDependency 的 Stage 才会有父 Stage, 也就是图中的虚线的位置！

所以我们只需要记住凡是涉及到 shuffle 的作业都会至少有两个 Stage，即 shuffle 前和 shuffle 后。

TaskScheduler 提交 Task

那我们接着看 submitMissingTasks 方法，下面是主体代码。

private def submitMissingTasks(stage: Stage, jobId: Int) {
    val myPending = pendingTasks.getOrElseUpdate(stage, new HashSet)
    myPending.clear()
    var tasks = ArrayBuffer[Task[_]]()
    if (stage.isShuffleMap) {
      // 这是shuffle stage的情况
      for (p <- 0 until stage.numPartitions if stage.outputLocs(p) == Nil) {
        val locs = getPreferredLocs(stage.rdd, p)
        tasks += new ShuffleMapTask(stage.id, stage.rdd, stage.shuffleDep.get, p, locs)
      }
    } else {
      // 这是final stage的情况
      val job = resultStageToJob(stage)
      for (id <- 0 until job.numPartitions if !job.finished(id)) {
        val partition = job.partitions(id)
        val locs = getPreferredLocs(stage.rdd, partition)
        tasks += new ResultTask(stage.id, stage.rdd, job.func, partition, locs, id)
      }
    }
    if (tasks.size > 0) {
      myPending ++= tasks
      taskScheduler.submitTasks(new TaskSet(tasks.toArray, stage.id, stage.newAttemptId(), stage.jobId, properties))
      stageToInfos(stage).submissionTime = Some(System.currentTimeMillis())
    } else {
      runningStages -= stage
    }
  }
复制代码

Task 也是有两类的，一种是 ShuffleMapTask，一种是 ResultTask，我们需要注意这两种 Task 的 runTask 方法。最后 Task 是通过 taskScheduler.submitTasks 来提交的。

我们找到 TaskSchedulerImpl 里面看这个方法。

override def submitTasks(taskSet: TaskSet) {
    val tasks = taskSet.tasksthis.synchronized {
      val manager = new TaskSetManager(this, taskSet, maxTaskFailures)
      activeTaskSets(taskSet.id) = manager
      schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
      hasReceivedTask = true
    }
    backend.reviveOffers()
  }
复制代码

调度器有两种模式，FIFO 和 FAIR，默认是 FIFO, 可以通过 spark.scheduler.mode 来设置，schedulableBuilder 也有相应的两种 FIFOSchedulableBuilder 和 FairSchedulableBuilder。

那 backend 是啥? 据说是为了给 TaskSchedulerImpl 提供插件式的调度服务的。

它是怎么实例化出来的，这里我们需要追溯回到 SparkContext 的 createTaskScheduler 方法，下面我直接把常用的 3 中类型的 TaskScheduler 给列出来了。

好，我们回到之前的代码上，schedulableBuilder.addTaskSetManager 比较简单，把作业集添加到调度器的队列当中。

我们接着看 backend 的 reviveOffers，里面只有一句话 driverActor ! ReviveOffers。真是头晕，搞那么多 Actor，只是为了接收消息。

照旧吧，找到它的 receive 方法，找到 ReviveOffers 这个 case，发现它调用了 makeOffers 方法，我们继续追杀！

def makeOffers() {
    launchTasks(scheduler.resourceOffers(executorHost.toArray.map {case (id, host) => new WorkerOffer(id, host, freeCores(id))}))
}
复制代码

从 executorHost 中随机抽出一些来给调度器，然后调度器返回 TaskDescription，executorHost 怎么来的，待会儿再说，我们接着看 resourceOffers 方法。

def resourceOffers(offers: Seq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {
    SparkEnv.set(sc.env)

    // 遍历worker提供的资源，更新executor相关的映射
    for (o <- offers) {
      executorIdToHost(o.executorId) = o.host
      if (!executorsByHost.contains(o.host)) {
        executorsByHost(o.host) = new HashSet[String]()
        executorAdded(o.executorId, o.host)
      }
    }

    // 从worker当中随机选出一些来，防止任务都堆在一个机器上
    val shuffledOffers = Random.shuffle(offers)
    // worker的task列表
    val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores))
    val availableCpus = shuffledOffers.map(o => o.cores).toArray
    val sortedTaskSets = rootPool.getSortedTaskSetQueue

    // 随机遍历抽出来的worker，通过TaskSetManager的resourceOffer，把本地性最高的Task分给Worker
    var launchedTask = false
    for (taskSet <- sortedTaskSets; maxLocality <- TaskLocality.values) {
      do {
        launchedTask = false
        for (i <- 0 until shuffledOffers.size) {
          val execId = shuffledOffers(i).executorId
          val host = shuffledOffers(i).host
          if (availableCpus(i) >= CPUS_PER_TASK) {
            // 把本地性最高的Task分给Worker
            for (task <- taskSet.resourceOffer(execId, host, maxLocality)) {
              tasks(i) += task
              val tid = task.taskId
              taskIdToTaskSetId(tid) = taskSet.taskSet.id
              taskIdToExecutorId(tid) = execId
              activeExecutorIds += execId
              executorsByHost(host) += execId
              availableCpus(i) -= CPUS_PER_TASK
              assert (availableCpus(i) >= 0)
              launchedTask = true
            }
          }
        }
      } while (launchedTask)
    }

    if (tasks.size > 0) {
      hasLaunchedTask = true
    }
    return tasks
  }
复制代码

resourceOffers 主要做了 3 件事：

• 从 Workers 里面随机抽出一些来执行任务。
• 通过 TaskSetManager 找出和 Worker 在一起的 Task，最后编译打包成 TaskDescription 返回。
• 将 Worker-->Array[TaskDescription] 的映射关系返回。

我们继续看 TaskSetManager 的 resourceOffer，看看它是怎么找到和 host 再起的 Task，并且包装成 TaskDescription。

通过查看代码，我发现之前我解释的和它具体实现的差别比较大，它所谓的本地性是根据当前的等待时间来确定的任务本地性的级别。

它的本地性主要是包括四类：PROCESS_LOCAL, NODE_LOCAL, RACK_LOCAL, ANY。

private def getAllowedLocalityLevel(curTime: Long): TaskLocality.TaskLocality = {
    while (curTime - lastLaunchTime >= localityWaits(currentLocalityIndex) &&
        currentLocalityIndex < myLocalityLevels.length - 1)
    {
      // 成立条件是当前时间-上次发布任务的时间 > 当前本地性级别的，条件成立就跳到下一个级别
      lastLaunchTime += localityWaits(currentLocalityIndex)
      currentLocalityIndex += 1
    }
    myLocalityLevels(currentLocalityIndex)
  }
复制代码

等待时间是可以通过参数去设置的，具体的自己查下面的代码。

private def getLocalityWait(level: TaskLocality.TaskLocality): Long = {
    val defaultWait = conf.get("spark.locality.wait", "3000")
    level match {
      case TaskLocality.PROCESS_LOCAL =>
        conf.get("spark.locality.wait.process", defaultWait).toLong
      case TaskLocality.NODE_LOCAL =>
        conf.get("spark.locality.wait.node", defaultWait).toLong
      case TaskLocality.RACK_LOCAL =>
        conf.get("spark.locality.wait.rack", defaultWait).toLong
      case TaskLocality.ANY =>
        0L
    }
  }
复制代码

下面继续看 TaskSetManager 的 resourceOffer 的方法，通过 findTask 来从 Task 集合里面找到相应的 Task。

findTask(execId, host, allowedLocality) match {
        case Some((index, taskLocality)) => {
             val task = tasks(index)
             val serializedTask = Task.serializeWithDependencies(task, sched.sc.addedFiles, sched.sc.addedJars, ser)
            val timeTaken = clock.getTime() - startTime
            addRunningTask(taskId)
            val taskName = "task %s:%d".format(taskSet.id, index)
            sched.dagScheduler.taskStarted(task, info)
            return Some(new TaskDescription(taskId, execId, taskName, index, serializedTask))
        }
复制代码

它的 findTask 方法如下：

private def findTask(execId: String, host: String, locality: TaskLocality.Value)
    : Option[(Int, TaskLocality.Value)] =
  {
   // 同一个Executor，通过execId来查找相应的等待的task
    for (index <- findTaskFromList(execId, getPendingTasksForExecutor(execId))) {
      return Some((index, TaskLocality.PROCESS_LOCAL))
    }
   // 通过主机名找到相应的Task,不过比之前的多了一步判断
    if (TaskLocality.isAllowed(locality, TaskLocality.NODE_LOCAL)) {
      for (index <- findTaskFromList(execId, getPendingTasksForHost(host))) {
        return Some((index, TaskLocality.NODE_LOCAL))
      }
    }
  // 通过Rack的名称查找Task
    if (TaskLocality.isAllowed(locality, TaskLocality.RACK_LOCAL)) {
      for {
        rack <- sched.getRackForHost(host)
        index <- findTaskFromList(execId, getPendingTasksForRack(rack))
      } {
        return Some((index, TaskLocality.RACK_LOCAL))
      }
    }
   // 查找那些preferredLocations为空的，不指定在哪里执行的Task来执行
    for (index <- findTaskFromList(execId, pendingTasksWithNoPrefs)) {
      return Some((index, TaskLocality.PROCESS_LOCAL))
    }
  // 查找那些preferredLocations为空的，不指定在哪里执行的Task来执行
    if (TaskLocality.isAllowed(locality, TaskLocality.ANY)) {
      for (index <- findTaskFromList(execId, allPendingTasks)) {
        return Some((index, TaskLocality.ANY))
      }
    }
    // 最后没办法了，拖的时间太长了，只能启动推测执行了
    findSpeculativeTask(execId, host, locality)
  }
复制代码

从这个方面可以看得出来，Spark 对运行时间还是很注重的，等待的时间越长，它就可能越饥不择食，从 PROCESS_LOCAL 一直让步到 ANY，最后的最后，推测执行都用到了。

找到任务之后，它就调用 dagScheduler.taskStarted 方法，通知 dagScheduler 任务开始了，taskStarted 方法就不详细讲了，它触发 dagScheduler 的 BeginEvent 事件，里面只做了 2 件事：

1、检查 Task 序列化的大小，超过 100K 就警告。

2、提交等待的 Stage。

好，我们继续回到发布 Task 上面来，中间过程讲完了，我们应该是要回到 CoarseGrainedSchedulerBackend 的launchTasks 方法了。

def makeOffers() {
    launchTasks(scheduler.resourceOffers(executorHost.toArray.map {case (id, host) => new WorkerOffer(id, host, freeCores(id))}))
}
复制代码

它的方法体是：

def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
      for (task <- tasks.flatten) {
        freeCores(task.executorId) -= scheduler.CPUS_PER_TASK
        executorActor(task.executorId) ! LaunchTask(task)
      }
    }
复制代码

通过 executorId 找到相应的 executorActor，然后发送 LaunchTask 过去，一个 Task 占用一个 Cpu。

注册 Application

那这个 executorActor 是怎么来的呢？找呗，最后发现它是在 receive 方法里面接受到 RegisterExecutor 消息的时候注册的。通过搜索，我们找到 CoarseGrainedExecutorBackend 这个类，在它的 preStart 方法里面赫然找到了 driver ! RegisterExecutor(executorId, hostPort, cores) 带的这三个参数都是在初始化的时候传入的，那是谁实例化的它呢，再逆向搜索找到 SparkDeploySchedulerBackend！之前的 backend 一直都是它，我们看 reviveOffers 是在它的父类 CoarseGrainedSchedulerBackend 里面。

关系清楚了，在这个 backend 的 start 方法里面启动了一个 AppClient，AppClient 的其中一个参数 ApplicationDescription 就是封装的运行 CoarseGrainedExecutorBackend 的命令。AppClient 内部启动了一个 ClientActor，这个 ClientActor 启动之后，会尝试向 Master 发送一个指令 actor ! RegisterApplication(appDescription) 注册一个 Application。

别废话了，Ctrl +Shift + N 吧，定位到 Master 吧。

case RegisterApplication(description) => {
        val app = createApplication(description, sender)
        registerApplication(app)
        persistenceEngine.addApplication(app)
        sender ! RegisteredApplication(app.id, masterUrl)
        schedule()
    }
复制代码

它做了 5 件事：

1、createApplication 为这个 app 构建一个描述 App 数据结构的 ApplicationInfo。

2、注册该 Application，更新相应的映射关系，添加到等待队列里面。

3、用 persistenceEngine 持久化 Application 信息，默认是不保存的，另外还有两种方式，保存在文件或者 Zookeeper 当中。

4、通过发送方注册成功。

5、开始作业调度。

关于调度的问题，在第一章 《spark-submit 提交作业过程》 已经介绍过了，建议回去再看看，搞清楚 Application 和 Executor 之间的关系。

Application 一旦获得资源，Master 会发送 launchExecutor 指令给 Worker 去启动 Executor。

进到 Worker 里面搜索 LaunchExecutor。

  val manager = new ExecutorRunner(appId, execId, appDesc, cores_, memory_, self, workerId, host,
            appDesc.sparkHome.map(userSparkHome => new File(userSparkHome)).getOrElse(sparkHome), workDir, akkaUrl, ExecutorState.RUNNING)
  executors(appId + "/" + execId) = manager
  manager.start()
   coresUsed += cores_
   memoryUsed += memory_
   masterLock.synchronized {
      master ! ExecutorStateChanged(appId, execId, manager.state, None, None)
   }
复制代码

原来 ExecutorRunner 还不是传说中的 Executor，它内部是执行了 appDesc 内部的那个命令，启动了 CoarseGrainedExecutorBackend，它才是我们的真命天子 Executor。

启动之后 ExecutorRunner 报告 ExecutorStateChanged 事件给 Master。

Master 干了两件事：

1、转发给 Driver，这个 Driver 是之前注册 Application 的那个 AppClient

2、如果是 Executor 运行结束，从相应的映射关系里面删除

发布 Task

上面又花了那么多时间讲 Task 的运行环境 ExecutorRunner 是怎么注册，那我们还是回到我们的主题，Task 的发布。

发布任务是发送 LaunchTask 指令给 CoarseGrainedExecutorBackend，接受到指令之后，让它内部的 executor 来发布这个任务。

这里我们看一下 Executor 的 launchTask。

def launchTask(context: ExecutorBackend, taskId: Long, serializedTask: ByteBuffer) {
    val tr = new TaskRunner(context, taskId, serializedTask)
    runningTasks.put(taskId, tr)
    threadPool.execute(tr)
  }
复制代码

TaskRunner 是这里的重头戏啊！看它的 run 方法吧。

override def run() {
      // 准备工作若干...那天我们放学回家经过一片玉米地，以上省略一百字

      try {
        // 反序列化Task
        SparkEnv.set(env)
        Accumulators.clear()
        val (taskFiles, taskJars, taskBytes) = Task.deserializeWithDependencies(serializedTask)
        updateDependencies(taskFiles, taskJars)
        task = ser.deserialize[Task[Any]](taskBytes, Thread.currentThread.getContextClassLoader)

       // 命令为尝试运行，和hadoop的mapreduce作业是一致的 
        attemptedTask = Some(task)
        logDebug("Task " + taskId + "'s epoch is " + task.epoch)
        env.mapOutputTracker.updateEpoch(task.epoch)

        // 运行Task, 具体可以去看之前让大家关注的ResultTask和ShuffleMapTask
        taskStart = System.currentTimeMillis()
        val value = task.run(taskId.toInt)
        val taskFinish = System.currentTimeMillis()

     // 对结果进行序列化
        val resultSer = SparkEnv.get.serializer.newInstance()
        val beforeSerialization = System.currentTimeMillis()
        val valueBytes = resultSer.serialize(value)
        val afterSerialization = System.currentTimeMillis()
     // 更新任务的相关监控信息，会反映到监控页面上的
        for (m <- task.metrics) {
          m.hostname = Utils.localHostName()
          m.executorDeserializeTime = taskStart - startTime
          m.executorRunTime = taskFinish - taskStart
          m.jvmGCTime = gcTime - startGCTime
          m.resultSerializationTime = afterSerialization - beforeSerialization
        }

        val accumUpdates = Accumulators.values
     // 对结果进行再包装，包装完再进行序列化
        val directResult = new DirectTaskResult(valueBytes, accumUpdates, task.metrics.getOrElse(null))
        val serializedDirectResult = ser.serialize(directResult)
        // 如果中间结果的大小超过了spark.akka.frameSize（默认是10M）的大小，就要提升序列化级别了，超过内存的部分要保存到硬盘的
        val serializedResult = {
          if (serializedDirectResult.limit >= akkaFrameSize - 1024) {
            val blockId = TaskResultBlockId(taskId)
            env.blockManager.putBytes(blockId, serializedDirectResult, StorageLevel.MEMORY_AND_DISK_SER)
            ser.serialize(new IndirectTaskResult[Any](blockId))
          } else {
            serializedDirectResult
          }
        }
     // 返回结果
        execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult)
      } catch {
        // 这部分是错误处理，被我省略掉了，主要内容是通关相关负责人处理后事
      } finally {
        // 清理为ResultTask注册的shuffle内存，最后把task从正在运行的列表当中删除
        val shuffleMemoryMap = env.shuffleMemoryMap
        shuffleMemoryMap.synchronized {
          shuffleMemoryMap.remove(Thread.currentThread().getId)
        }
        runningTasks.remove(taskId)
      }
    }
  }
复制代码

以上代码被我这些了，但是建议大家看看注释吧。

最后结果是通过 statusUpdate 返回的。

override def statusUpdate(taskId: Long, state: TaskState, data: ByteBuffer) {
    driver ! StatusUpdate(executorId, taskId, state, data)
  }
复制代码

这回这个 Driver 又不是刚才那个 AppClient，而是它的家长 SparkDeploySchedulerBackend，是在 SparkDeploySchedulerBackend 的父类 CoarseGrainedSchedulerBackend 接受了这个 StatusUpdate 消息。

这关系真他娘够乱的。。

继续，Task 里面走的是 TaskSchedulerImpl 这个方法。

scheduler.statusUpdate(taskId, state, data.value)
复制代码

到这里，一个 Task 就运行结束了，后面就不再扩展了，作业运行这块是 Spark 的核心，再扩展基本就能写出来一本书了，限于文章篇幅，这里就不再深究了。

以上的过程应该是和下面的图一致的。

看完这篇文章，估计大家会云里雾里的，在下一章《作业生命周期》会把刚才描述的整个过程重新梳理出来，便于大家记忆，敬请期待！