Spark 源码分析（九）: Task 启动

前面已经分析到了 DAGScheduler 对 stage 划分，并对 Task 的最佳位置进行计算以后，经过调用 taskScheduler 的 submitTasks 方法，将每一个 stage 的 taskSet 进行提交。

在 taskScheduler 的 submitTasks 方法中会为每一个 taskSet 建立一个 TaskSetManager，用于管理 taskSet。而后向调度池中添加该 TaskSetManager，最后会调用 backend.reviveOffers() 方法为 task 分配资源。

override def submitTasks(taskSet: TaskSet) {
  	//获取 taskSet 中的 task
    val tasks = taskSet.tasks logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks") this.synchronized {
      // 为每一个 taskSet 建立一个 TaskSetManager
      val manager = createTaskSetManager(taskSet, maxTaskFailures)
      // 拿到 stage 的 id
      val stage = taskSet.stageId
      // 建立一个 HashMap ，用来存储 stage 对应的 TaskSetManager
      val stageTaskSets =
        taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
      // 将上面建立的 taskSetManager 存入 map 中
      stageTaskSets(taskSet.stageAttemptId) = manager
      val conflictingTaskSet = stageTaskSets.exists { case (_, ts) =>
        ts.taskSet != taskSet && !ts.isZombie
      }
      if (conflictingTaskSet) {
        throw new IllegalStateException(s"more than one active taskSet for stage $stage:" +
          s" ${stageTaskSets.toSeq.map{_._2.taskSet.id}.mkString(",")}")
      }
      // 向调度池中添加刚才建立的 TaskSetManager
      schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)

      // 判断程序是否为 local 模式，而且 TaskSchedulerImpl 没有收到 Task
      if (!isLocal && !hasReceivedTask) {
        // 建立一个定时器，经过指定时间检查 TaskSchedulerImpl 的饥饿状况
        starvationTimer.scheduleAtFixedRate(new TimerTask() {
          override def run() {
            // 若是 TaskSchedulerImpl 已经安排执行了 Task，则取消定时器
            if (!hasLaunchedTask) {
              logWarning("Initial job has not accepted any resources; " +
                "check your cluster UI to ensure that workers are registered " +
                "and have sufficient resources")
            } else {
              this.cancel()
            }
          }
        }, STARVATION_TIMEOUT_MS, STARVATION_TIMEOUT_MS)
      }
      // 标记已经接收到 Task
      hasReceivedTask = true
    }
  	// 给 Task 分配资源
    backend.reviveOffers()
  }
复制代码

下面主要看 backend.reviveOffers() 这个方法，在提交模式是 standalone 模式下，其实是调用 StandaloneSchedulerBackend 的 reviveOffers 方法，实则调用的是其父类 CoarseGrainedSchedulerBackend 的 reviveOffers 方法，这个方法是向 driverEndpoint 发送一个 ReviveOffers 消息。

override def reviveOffers() {
  	// 向 driverEndpoint 发送 ReviveOffers 消息
    driverEndpoint.send(ReviveOffers)
  }
复制代码

DriverEndpoint 收到信息后会调用 makeOffers 方法：

case ReviveOffers =>
        makeOffers()
复制代码

makeOffers 方法内部会将 application 全部可用的 executor 封装成一个 workOffers，每一个 workOffers 内部封装了每一个 executor 的资源数量。

而后调用 taskScheduler 的 resourceOffers 从上面封装的 workOffers 信息为每一个 task 分配合适的 executor。

最后调用 launchTasks 启动 task。

private def makeOffers() {
      // 过滤出可用的 executor
      val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
      // 将这些 executor 封装成 workOffers
      val workOffers = activeExecutors.map { case (id, executorData) =>
        new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
      }.toIndexedSeq
      // 给每一个 task 分配 executor，而后调用 launchTasks 启动这些 task
      launchTasks(scheduler.resourceOffers(workOffers))
    }
复制代码

下面看一下 launchTasks 这个方法。

这个方法主要作了这些操做：

1，遍历每一个 task，而后将每一个 task 信息序列化。

2，判断序列化后的 task 信息，若是大于 rpc 发送消息的最大值，则中止，建议调整 rpc 的 maxRpcMessageSize，若是小于 rpc 发送消息的最大值，则找到 task 对应的 executor，而后更新该 executor 对应的一些内存资源信息。

3，向 executor 发送 LaunchTask 消息。

private def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
  		// 遍历全部的 task
      for (task <- tasks.flatten) {
        // 序列化 task 信息
        val serializedTask = ser.serialize(task)
        // 判断序列化后的 task 信息是否大于 rpc 可以传送的最大信息量
        if (serializedTask.limit >= maxRpcMessageSize) {
          scheduler.taskIdToTaskSetManager.get(task.taskId).foreach { taskSetMgr =>
            try {
              var msg = "Serialized task %s:%d was %d bytes, which exceeds max allowed: " +
                "spark.rpc.message.maxSize (%d bytes). Consider increasing " +
                "spark.rpc.message.maxSize or using broadcast variables for large values."
              msg = msg.format(task.taskId, task.index, serializedTask.limit, maxRpcMessageSize)
              taskSetMgr.abort(msg)
            } catch {
              case e: Exception => logError("Exception in error callback", e)
            }
          }
        }
        else {
          // 找到对应的 executor
          val executorData = executorDataMap(task.executorId)
          // 更新 executor 的资源信息
          executorData.freeCores -= scheduler.CPUS_PER_TASK logDebug(s"Launching task ${task.taskId} on executor id: ${task.executorId} hostname: " + s"${executorData.executorHost}.") // 向 executor 发送 LaunchTask 消息 executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask))) } } } 复制代码

下面就看 Executor 收到消息后作了哪些操做，这里 executorData.executorEndpoint 实际上就是在建立 Executor 守护进程时候建立的那个 CoarseGrainedExecutorBackend。

因此找到 CoarseGrainedExecutorBackend 处理接收到 LaunchTask 消息后作了哪些操做。

首先他会判断当前的 executor 是否是为空，若是不为空就会反序列化 task 的信息，而后调用 executor 的 launchTask 方法。

case LaunchTask(data) =>
  		// 判断当前 executor 是否是空
      if (executor == null) {
        exitExecutor(1, "Received LaunchTask command but executor was null")
      } else {
        // 反序列化 task 的信息
        val taskDesc = ser.deserialize[TaskDescription](data.value)
        logInfo("Got assigned task " + taskDesc.taskId)
        // 调用 executor 的 lauchTask 方法
        executor.launchTask(this, taskId = taskDesc.taskId, attemptNumber = taskDesc.attemptNumber,
          taskDesc.name, taskDesc.serializedTask)
      }
复制代码

下面就看下 executor 的 launchTask 作了哪些操做。

首先 executor 会为每一个 task 建立一个 TaskRunner，而后会会将 task 添加到 runningTasks 的集合中，并标记其为运行状态，最后将 taskRunner 放到一个线程池中执行。

def launchTask( context: ExecutorBackend, taskId: Long, attemptNumber: Int, taskName: String, serializedTask: ByteBuffer): Unit = {
  	// 建立 TaskRunner
    val tr = new TaskRunner(context, taskId = taskId, attemptNumber = attemptNumber, taskName,
      serializedTask)
    runningTasks.put(taskId, tr)
    // 将 taskRunner 放到线程池中执行
    threadPool.execute(tr)
  }
复制代码

在 taskRunner 的 run 方法中回去执行每一个 task，并会输出一系列的日志。task 运行完成后回向 driver 发送消息，driver 会更新 executor 的一些资源数据，并标记 task 已完成。

至此 task 启动完成。