Spark源码分析 -- TaskScheduler

  private var taskScheduler: TaskScheduler = {
      case SPARK_REGEX(sparkUrl) =>
        val scheduler = new ClusterScheduler(this)  // 创建ClusterScheduler
        val backend = new SparkDeploySchedulerBackend(scheduler, this, sparkUrl, appName) // 创建SparkDeploySchedulerBackend
        scheduler.initialize(backend)
        scheduler
  } 

/**
 * Low-level task scheduler interface, implemented by both ClusterScheduler and LocalScheduler.
 * These schedulers get sets of tasks submitted to them from the DAGScheduler for each stage,
 * and are responsible for sending the tasks to the cluster, running them, retrying if there
 * are failures, and mitigating stragglers. They return events to the DAGScheduler through
 * the TaskSchedulerListener interface.
 */
private[spark] trait TaskScheduler {
  def rootPool: Pool
  def schedulingMode: SchedulingMode
  def start(): Unit // 启动
  def postStartHook() { }
  def stop(): Unit
  // Submit a sequence of tasks to run.
  def submitTasks(taskSet: TaskSet): Unit // 核心, 提交taskset的接口
  // Set a listener for upcalls. This is guaranteed to be set before submitTasks is called.
  def setListener(listener: TaskSchedulerListener): Unit // TaskScheduler会使用这个listener来汇报当前task的运行状况,会注册DAGScheduler
  // Get the default level of parallelism to use in the cluster, as a hint for sizing jobs.
  def defaultParallelism(): Int
}

private[spark] class ClusterScheduler(val sc: SparkContext) extends TaskScheduler with Logging
{
  var listener: TaskSchedulerListener = null
  var backend: SchedulerBackend = null
  val mapOutputTracker = SparkEnv.get.mapOutputTracker
  var schedulableBuilder: SchedulableBuilder = null
  var rootPool: Pool = null
  // default scheduler is FIFO
  val schedulingMode: SchedulingMode = SchedulingMode.withName(
    System.getProperty("spark.scheduler.mode", "FIFO"))

  def initialize(context: SchedulerBackend) {
    backend = context    // 初始化SchedulerBackend 
    // temporarily set rootPool name to empty
    rootPool = new Pool("", schedulingMode, 0, 0) // 创建Schedulable tree的root pool
    schedulableBuilder = { // 用schedulableBuilder初始化Schedulable tree
      schedulingMode match {
        case SchedulingMode.FIFO =>
          new FIFOSchedulableBuilder(rootPool)
        case SchedulingMode.FAIR =>
          new FairSchedulableBuilder(rootPool)
      }
    }
    schedulableBuilder.buildPools()
  }

  override def start() {
    backend.start() // 启动SchedulerBackend 
  }

  override def submitTasks(taskSet: TaskSet) {
    val tasks = taskSet.tasks
    logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
    this.synchronized {
      val manager = new ClusterTaskSetManager(this, taskSet)
      activeTaskSets(taskSet.id) = manager
      schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties) // 将TaskSetManager加到Schedulable tree等待被调度执行
      taskSetTaskIds(taskSet.id) = new util.HashSet[Long]()
      backend.reviveOffers() // 调用SchedulerBackend的reviveOffers, 其实就是往DriverActor发送reviveOffers事件
  }

  /**
   * Called by cluster manager to offer resources on slaves. We respond by asking our active task
   * sets for tasks in order of priority. We fill each node with tasks in a round-robin manner so
   * that tasks are balanced across the cluster.
   */

  // 根据当前可用的worker offers, 分配tasks
  def resourceOffers(offers: Seq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {
    SparkEnv.set(sc.env)

    // Build a list of tasks to assign to each worker
    val tasks = offers.map(o => new ArrayBuffer[TaskDescription](o.cores)) // 每个core可以分配一个task,所以对每个offer生成length为cores数目的ArrayBuffer
    val availableCpus = offers.map(o => o.cores).toArray  // 每个work可用的core数目的array
    val sortedTaskSets = rootPool.getSortedTaskSetQueue() // 得到根据schedule算法排序后的TaskSetManager列表
    // Take each TaskSet in our scheduling order, and then offer it each node in increasing order
    // of locality levels so that it gets a chance to launch local tasks on all of them.
    var launchedTask = false
    for (taskSet <- sortedTaskSets; maxLocality <- TaskLocality.values) { // 嵌套, 遍历sortedTaskSets, 并对每个taskSet遍历所有TaskLocality
      do {
        launchedTask = false
        for (i <- 0 until offers.size) { // 遍历每个offer, 试图在当前的taskset和当前的locality上找到合适的task
          val execId = offers(i).executorId
          val host = offers(i).host
          for (task <- taskSet.resourceOffer(execId, host, availableCpus(i), maxLocality)) { // 每次只会返回最多一个task
            tasks(i) += task
            val tid = task.taskId
            taskIdToTaskSetId(tid) = taskSet.taskSet.id
            taskSetTaskIds(taskSet.taskSet.id) += tid
            taskIdToExecutorId(tid) = execId
            activeExecutorIds += execId
            executorsByHost(host) += execId
            availableCpus(i) –= 1 // 分配一个task, 所以availableCpus - 1
            launchedTask = true
          }
        }
      } while (launchedTask) // 找到,就继续在这个locality上找task, 否则放宽到下个locality,或下个taskset
    }

    if (tasks.size > 0) {
      hasLaunchedTask = true
    }
    return tasks
  }

}

本文章摘自博客园，原文发布日期： 2014-01-03