TaskScheduler源码解读

原创

幽鸿

发布于 2020-05-04 12:29:18

50800

代码可运行

文章被收录于专栏：大数据-数据人生大数据-数据人生

运行总次数：0

代码可运行

在DAGScheduler中提交task方法入口：

taskScheduler.submitTasks(new TaskSet(
  tasks.toArray, stage.id, stage.latestInfo.attemptId, jobId, properties))

这里调用了taskScheduler接口，我们打开TaskScheduler trait，trait在scala里就是接口，在IDEA中查看实现的类，使用快捷键：ctrl+H，或者直接使用快捷键：ctrl + alt +B查看实现

查看submitTasks方法实现：

override def submitTasks(taskSet: TaskSet) {
  val tasks = taskSet.tasks
  logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
  this.synchronized {
   //给每一个TaskSet创建一个TaskSetManager
    val manager = createTaskSetManager(taskSet, maxTaskFailures)
    val stage = taskSet.stageId
    val stageTaskSets =
      taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
    //将manager进行内存缓存,manager负责对task进行跟踪管理，包括重试
    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(",")}")
    }
    schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)

    if (!isLocal && !hasReceivedTask) {
      starvationTimer.scheduleAtFixedRate(new TimerTask() {
        override def run() {
          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)
    }
    hasReceivedTask = true
  }
  //这里的backend就是SparkContext创建好的SparkDeploySchedulerBackend
  //这里backend负责创建AppClient,向Master注册Application
  //参见SparkContext代码createTaskScheduler方法
  backend.reviveOffers()
}