前面几章我们重点讲述了spark的的原理，sparkContext的初使化，spark主备切换，master的注册等。其中我们分析源码时，不管是driver,还是application在注册Master时都，在最后都有这样的一个方法调用：

 ...
 schedule()

这个方法其实就是资源调度算法（这个方法太重要了，也是核心中的核心）
下面我们就来深入分析这个方法。

 private def schedule() {
  #当master不是alive时，直接reuturn
  #也就是说Standby是不参与资源调度的
    if (state != RecoveryState.ALIVE) { return }

//Random.shuffle作用就是把集合随机打乱
//取出workers中所有之前注册的worker,进行过滤，必须 状态 是Alive的worker
//把worker随机的打乱
 val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(
_.state == WorkerState.ALIVE))
    val numWorkersAlive = shuffledAliveWorkers.size

为什么要调度driver,大家想一下什么时候会注册driver，并且导致driver被调度
其实只有在模式是yarn-cluster提交后，才会注册driver,因为standalone与yarn-client
都会在本地启动dirver,而不会来注册driver,就更不可能被master来调度
所以说下面的这个for只会运行在yarn-cluster模式下提交下。

 for (driver <- waitingDrivers.toList) { // iterate over a copy of waitingDrivers
      // We assign workers to each waiting driver in a round-robin fashion. For each driver, we
      // start from the last worker that was assigned a driver, and continue onwards until we have
      // explored all alive workers.
      var launched = false
      var numWorkersVisited = 0
      
      /**while中的条件，当还有活着的worker没有被遍历到，就继续遍历
       * 而且这个driver在这个worker中还没有启动,launched=false
       */
      while (numWorkersVisited < numWorkersAlive && !launched) {
        val worker = shuffledAliveWorkers(curPos)
        numWorkersVisited += 1
        /**
         * 如果这个worker的空闲内存容量 大于等于driver所需的内存
         * 而且worker空间的CPU大于等于driver所需的CPU数量
         */
        if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
         //启动driver
          launchDriver(worker, driver)
          //把此driver从waitingDrivers中去掉
          waitingDrivers -= driver
          launched = true
        }
        //将指针指向下一个worker
        curPos = (curPos + 1) % numWorkersAlive
      }
    }

上面就是对Drvier的调度。
接着我们看对Application的资源调度
/**
* Application的调度机制（核心之核心 ）
* 两种算法：一种是spreadOutApps(默认),另一种是非spreadOutApps
*
* 通过个算法，其实 会将每个application,要启动的executor都平均分配 到每个worker上
* 比如有20cpu core,有10个worker,那么实际会遍历两遍，每次循环，每个worker分配一个core
* 最后每个worker分配了两个core
*/

  if (spreadOutApps) {
          for (app <- waitingApps if app.coresLeft > 0) {
        //从workerk中，过滤出状态是ALIVE的
        val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE)
          .filter(canUse(app, _)).sortBy(_.coresFree).reverse
        val numUsable = usableWorkers.length
        //创建一个空数组，存储了要分配的每个worker的cpu
        val assigned = new Array[Int](numUsable) // Number of cores to give on each node
        //获取到底可以分配多少个cpu,取application所需的cpu数量与worker可用的cpu数量的最小值
        var toAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)
        var pos = 0
        //只要有还有要分配的cpu没有分配完就while
        while (toAssign > 0) {
          //如果可用的cpu大于已经分配的cpu数量，其实就是还有可用的cpu
          if (usableWorkers(pos).coresFree - assigned(pos) > 0) {
            //将要分配的cpu数量减一
            toAssign -= 1
            //给这个worker分配的cpu加1
            assigned(pos) += 1
          }
          //指定指向下一个worker
          pos = (pos + 1) % numUsable
        }
        // Now that we've decided how many cores to give on each node, let's actually give them
       //给每个worker分配了application需要的cpu core后
        for (pos <- 0 until numUsable) {
          //判断这个worker已经分配了core
          if (assigned(pos) > 0) {
            //创建了ExecutorDesc对象，封装了executor的信息
            //将这个executor添加到application缓存区
            val exec = app.addExecutor(usableWorkers(pos), assigned(pos))
            //在worker上启动Executor
            launchExecutor(usableWorkers(pos), exec)
            //将application的状态为RUNNING
            app.state = ApplicationState.RUNNING
           }
        }
      }
    }

关于这种调度算法总结：

我们之前说了在spark-submit中指定了多少个executor,每个execuotr需要多少个cpu core 实际上基本这个机制，最后，executor的实际数量，每个executor需要的core,可能与配置不一样 因为这里我们是基于总的cpu来分配的，就是说比如，我们配置了需要三个executor来启动application, 每个executor需要三个core，那么就总需9个core,其实在这种算法中，如果我们有9个worker,会给每个 worker分配一个core,然后给每个worker启动一个executor.最后其实是启动了9个executor,每个 executor有一个core

第二种调度算法：

这种算法与上面的正好相反，每个application,都尽可能少的分配到worker上去， 比如总共有10个worker,每个有10个core application总共要分配20个core,那么只会分配到两个worker上，每个worker都占满了这10个core那么其它的application只能分配另外的worker上去了。 所以我们在spark-submit中配置了要10个executor,每个execuotr需要2个core 那么共需要20个core,但这种算法中，其实只会启动两个executor，每个executor有10个core

//遍历worker,并且状态是ALIVE，还有空闲的cpu的worker
      for (worker <- workers if worker.coresFree > 0 && worker.state == WorkerState.ALIVE) {
       //遍历application,并且还有需要分配的core的applicztion
        for (app <- waitingApps if app.coresLeft > 0) {
          //判断当前这个worker可以被application使用
          if (canUse(app, worker)) {
            //取worker可用的cpu数量与application要分配的cpu数量的最小值
            val coresToUse = math.min(worker.coresFree, app.coresLeft)
            //如果小于0，说明没有core可分了
            if (coresToUse > 0) {
              val exec = app.addExecutor(worker, coresToUse)
                      //在worker上启动executor
              launchExecutor(worker, exec)
              //设置application的状态是
              app.state = ApplicationState.RUNNING
            }
          }

其中里面有一个非常重要的方法：

launchExecutor(worker, exec)

def launchExecutor(worker: WorkerInfo, exec: ExecutorDesc) {
    logInfo("Launching executor " + exec.fullId + " on worker " + worker.id)
    //将executor加入worker缓存
    worker.addExecutor(exec)
    //向worker的actor发送LaunchExecutor ,在worker中启动executor
    worker.actor ! LaunchExecutor(masterUrl,
      exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory)
    exec.application.driver ! ExecutorAdded(
      exec.id, worker.id, worker.hostPort, exec.cores, exec.memory)
  }

这个方法里的
worker.actor ! LaunchExecutor(masterUrl,
exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory)
正是向worker发送这个消息（封闭了akka的消息通信） ，在worker端调用这个方法来启动Executor(关于worker如何启动Eexecutor与Application，我们在下一节会详细的剖析)

本章中每一个字（包括源码注解）都是作者敲出来的，你感觉有用，帮点击’喜欢’