本文為 Spark 2.0 版本的源碼分析，其他版本可能會(huì)有所不同
TaskScheduler 作為資源調(diào)度器的一個(gè)重要職責(zé)就在：
- 集群可用資源發(fā)生變化（比如有新增的 executor，有 executor lost 等）
- 有新的 task 提交
- 有 task 結(jié)束
- 處理 Speculatable task
等時(shí)機(jī)把處于等待狀態(tài)的 tasks 分配給有空閑資源的 executors，那么這個(gè) “把 task 分配給 executor” 的過(guò)程具體是怎樣的呢？這就是本文要探討的內(nèi)容，將通過(guò)以下四小節(jié)來(lái)進(jìn)行剖析：
- 打散可用的 executors
- 對(duì)所有處于等待狀態(tài)的 taskSet 進(jìn)行排序
- 根據(jù)是否有新增的 executor 來(lái)決定是否更新各個(gè) taskSet 的可用本地性集合
- 結(jié)合 taskSets 的排序及本地性集合將 tasks 分配給 executors
打散可用的 executors
“把 task 分配給 executor” 這一過(guò)程是在函數(shù) TaskSchedulerImpl#resourceOffers(offers: Seq[WorkerOffer]): Seq[Seq[TaskDescription]] 中完成的：
- 傳入?yún)?shù)
offers: Seq[WorkerOffer]為集群中所有 activeExecutors 一一對(duì)應(yīng)，一個(gè) WorkerOffer 包含的信息包括：executorId、executorHost及該 executor 空閑的 cores - 返回值類型為
Seq[Seq[TaskDescription]]，其每一個(gè)元素（即Seq[TaskDescription]類型）為經(jīng)過(guò)該函數(shù)的調(diào)度分配給下標(biāo)相同的 offers 元素對(duì)應(yīng)的 executor 的 tasks
TaskSchedulerImpl#resourceOffers 第一階段做的事可用下圖表示：
在該函數(shù)每次被調(diào)用之時(shí)，通過(guò)隨機(jī)的方式打亂所有 workerOffers（一個(gè) workerOffer 對(duì)應(yīng)一個(gè)active executor），之后會(huì)根據(jù)這打亂后的順序給 executor 分配 task，這樣做就能避免只將 tasks 分配給少數(shù)幾個(gè) executors 從而達(dá)到使集群各節(jié)點(diǎn)壓力平均的目的。
除了打散 workerOffers，還為每個(gè) workerOffer 創(chuàng)建了一個(gè)可變的 TaskDescription 數(shù)組從而組成了一個(gè)二維數(shù)組 tasks，用于保存之后的操作中分配給各個(gè) executor 的 tasks。
對(duì)所有處于等待狀態(tài)的 taskSet 進(jìn)行排序
排序的目的是為了讓優(yōu)先級(jí)更高的 taskSet 所包含的 task 更優(yōu)先的被調(diào)度執(zhí)行，所執(zhí)行的操作是：
val sortedTaskSets: ArrayBuffer[TaskSetManager] = rootPool.getSortedTaskSetQueue
其中，sortedTaskSets 是排序后得到的 TaskSetManager 數(shù)組，下標(biāo)越小表示優(yōu)先級(jí)越高，也就越優(yōu)先被調(diào)度。而 rootPool 是 Pool 類型，它是 Standalone 模式下的對(duì)列，支持兩種調(diào)度模式，分別是：
- SchedulingMode.FIFO：FIFO 模式，先進(jìn)先出
- SchedulingMode.FAIR：公平模式，會(huì)考慮各個(gè)對(duì)列資源的使用情況
更具體的分析，請(qǐng)移步Pool-Standalone模式下的隊(duì)列，這篇文章對(duì)兩種調(diào)度方式以及如何排序做做了十分詳細(xì)的說(shuō)明
根據(jù)是否有新增的 executor 來(lái)決定是否更新各個(gè) taskSet 的可用本地性集合
關(guān)于更新 taskSet 的可用本地性集合，這里值進(jìn)行簡(jiǎn)單說(shuō)明，更多內(nèi)容請(qǐng)移步 Spark的位置優(yōu)先: TaskSetManager 的有效 Locality Levels
- 若 taskSet 中有 task 的 partition 是存儲(chǔ)在 executor 內(nèi)存中的且對(duì)應(yīng) executor alive，那么該 taskSet 的最佳本地性為
PROCESS_LOCAL,可用本地性集合包括PROCESS_LOCAL及所有本地性比PROCESS_LOCAL查的，也就是該集合包括PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY - 若 taskSet 中沒(méi)有 task 的 partition 是存儲(chǔ)在 executor 內(nèi)存中的，但存在 partition 是存儲(chǔ)在某個(gè)節(jié)點(diǎn)磁盤上的且對(duì)應(yīng)節(jié)點(diǎn) alive ，那么該 taskSet 的最佳本地性為
NODE_LOCAL,可用本地性集合包括NODE_LOCAL及所有本地性比NODE_LOCAL查的，也就是該集合包括NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY - 以此類推，可用本地性集合包含 taskSet 中的 tasks 所擁有的最佳本地性及所有比該本地性差的本地性
這個(gè)可用本地性集合會(huì)在后面的將 task 分配給 executor 起關(guān)鍵作用
結(jié)合 taskSets 的排序及本地性集合將 tasks 分配給 executors
這一步的實(shí)現(xiàn)代碼如下：
for (taskSet <- sortedTaskSets; maxLocality <- taskSet.myLocalityLevels) {
do {
launchedTask = resourceOfferSingleTaskSet(
taskSet, maxLocality, shuffledOffers, availableCpus, tasks)
} while (launchedTask)
}
含義是根據(jù) sortedTaskSets 的順序依次遍歷其每一個(gè) taskSetManager，
從該 taskSetManager 的本地性從高到低依次調(diào)用 TaskSchedulerImpl#resourceOfferSingleTaskSet，流程如下圖：
以上，就完成了分配 tasks 給 executors 的流程分析，細(xì)節(jié)比較多，涉及的知識(shí)點(diǎn)也比較多，需要擴(kuò)展閱讀文中給出的另幾個(gè)文章，最后給出一個(gè)整體的流程圖方便理解
