sparkContext初始化後會註冊Application，然後會呼叫schedule方法，如何為Application在worker上啟動Executor，Executor啟動後，DAGScheduler和TaskScheduler才能分配task給Executor來進行計算。所以schedule是把整個流程竄起來的重點。

private def schedule(): Unit = {
  //standby master是不會進行Application等資源排程的
  if (state != RecoveryState.ALIVE) {
    return
}
  // Drivers take strict precedence over executors
  //第一行重要程式碼，Random.shuffle的原理，將集合的元素隨機大量
    取出workers中所有之前註冊上來的workers，進行過濾，必須是狀態ALIVE的worker
    對狀態為ALIVE的worker，呼叫Random的shuffle方法進行隨機打亂
 
val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(_.state == WorkerState.ALIVE))
  val numWorkersAlive = shuffledAliveWorkers.size
  var curPos = 0
  //首先排程Driver，什麼情況下會註冊Driver並且導致Driver被排程，其實只有用yarn-cluster模式提交
    才會。因為standalone和yarn-client模式，都是在本地直接啟動Driver，而不會來註冊Driver，更不可能排程Driver
    遍歷waitingDrivers ArrayBuffer
 
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
    //只要有活著的worker沒有遍歷到，並且driver還沒有被啟動，也就是launched為false
while (numWorkersVisited < numWorkersAlive && !launched) {
      val worker = shuffledAliveWorkers(curPos)
      numWorkersVisited += 1
    //如果當前這個worker的空閒記憶體和cpu數量大於等於Driver需要的
if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
        //啟動Driver
        launchDriver(worker, driver)
        //並且將driver從ArrayBuffer中移除
        waitingDrivers -= driver
        launched = true
}
      //將指標指向下一個worker
      curPos = (curPos + 1) % numWorkersAlive
    }
  }
  startExecutorsOnWorkers()
}

private def launchDriver(worker: WorkerInfo, driver: DriverInfo) {
  logInfo("Launching driver " + driver.id + " on worker " + worker.id)
  //將driver加入worker記憶體的快取結構
    將worker內使用的記憶體和cpu數量，都加上driver需要的記憶體和cpu數量
  worker.addDriver(driver)
  //同時把worker也加入到driver的快取結構中
  driver.worker = Some(worker)
  //然後呼叫worker的RpcEndpoint，給它傳送LaunchDriver訊息，讓worker來啟動Driver
  worker.endpoint.send(LaunchDriver(driver.id, driver.desc))
  //將driver的狀態設定為Running
  driver.state = DriverState.RUNNING
}

 Application的排程機制（核心之核心 ）
 兩種演算法：一種是spreadOutApps(預設),另一種是非spreadOutApps

 通過spreadOutApps(預設)演算法，其實 會將每個application,要啟動的executor都平均分配 到每個worker上
 比如有20cpu core,有10個worker,那麼實際會遍歷兩遍，每次迴圈，每個worker分配一個core
 最後每個worker分配了兩個core

非spreadOutApps演算法與上面的正好相反，每個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

//這個方法就是真正啟動executor的方法,在執行這個方法之前，會呼叫一些其他的驗證方法，得到一個結果集合
//assignedCores，這個集合計算出了每個一個work上能分配幾個core。通過這個結果，就能知道啟動幾個executor
private def allocateWorkerResourceToExecutors(
        app: ApplicationInfo,
assignedCores: Int,
coresPerExecutor: Option[Int],
worker: WorkerInfo): Unit = {
    // If the number of cores per executor is specified, we divide the cores assigned
    // to this worker evenly among the executors with no remainder.
    // Otherwise, we launch a single executor that grabs all the assignedCores on this worker.
    //在迴圈的當前worker裡，要啟動exec的個數 （該worker的總數core / 每個exec需要的core = exec個數,如果沒配置每個exec所需core，則預設為1）
val numExecutors = coresPerExecutor.map { assignedCores / _ }.getOrElse(1)
    //如果沒配置每個exec所需core,直接在這個把分配給這個worker的所有core全部用來啟動這個exec，否者按照配置的來
val coresToAssign = coresPerExecutor.getOrElse(assignedCores)
    for (i <- 1 to numExecutors) {
        val exec = app.addExecutor(worker, coresToAssign)
        launchExecutor(worker, exec)
        app.state = ApplicationState.RUNNING
    }
}
/**
 總結:
 提交任務時指定每個exec分配2個core,啟動3個executor
 那麼在spark會用預設演算法spreadOutApps,平均給每個worker分配資源的情況下
 先計算出總core數  2*3 = 6
 然後給某三個worker一個分配1個exec（assignedCores集合裡存兩個Int：2,2,2 代表三個worker分別分配2個core）
 然後公式 assignedCores(該worker啟動exec所需core總數) / coresPerExecutor(配置的每個exec啟動core個數) = （2/2=1）（該worker啟動的exe個數）
 然後公式 coresPerExecutor.getOrElse(assignedCores) 到底要啟動幾個core（2）
 最後:得到了要啟動2個core，得到了要啟動exec的個數,就迴圈exec個數來分別啟動2個core