Spark的調度策略詳解

Spark的調度策略

Spark目前有兩種調度策略，一種是FIFO即先來先得，另一種是FAIR即公平策略。所謂的調度策略就是對待調度的對象進行排序，按照優先級來進行調度。調度的排序接口如下所示，就是對兩個可調度的對象進行比較。

private[spark] trait SchedulingAlgorithm { def comparator(s1: Schedulable, s2: Schedulable): Boolean} 

其實現類為FIFOSchedulingAlgorithm、FairSchedulingAlgorithm

/** * FIFO排序的實現，主要因素是優先級、其次是對應的Stage * 優先級高的在前面，優先級相同，則靠前的stage優先 */ private[spark] class FIFOSchedulingAlgorithm extends SchedulingAlgorithm { override def comparator(s1: Schedulable, s2: Schedulable): Boolean = { //一般來說優先級越小優先級越高 val priority1 = s1.priority val priority2 = s2.priority var res = math.signum(priority1 - priority2) if (res == 0) { //如果優先級相同，那麼Stage靠前的優先 val stageId1 = s1.stageId val stageId2 = s2.stageId        res = math.signum(stageId1 - stageId2)    } if (res < 0) { true } else { false } }} 

注：
可以根據自己對優先級的定義重寫這個比較方法，但有一點注意，就是如果優先級和Stage都相同，那麼默認後來居上

private[spark] class FairSchedulingAlgorithm extends SchedulingAlgorithm { override def comparator(s1: Schedulable, s2: Schedulable): Boolean = { //最小共享,可以理解為執行需要的最小資源即CPU核數，其他相同時，所需最小核數小的優先調度 val minShare1 = s1.minShare val minShare2 = s2.minShare //運行的任務的數量 val runningTasks1 = s1.runningTasks val runningTasks2 = s2.runningTasks //是否有處於挨餓狀態的任務，看可分配的核數是否少於任務數，如果資源不夠用，那麼處於挨餓狀態 val s1Needy = runningTasks1 < minShare1 val s2Needy = runningTasks2 < minShare2 //最小資源占用比例，這裡可以理解為偏向任務較輕的  val minShareRatio1 = runningTasks1.toDouble / math.max(minShare1, 1.0).toDouble val minShareRatio2 = runningTasks2.toDouble / math.max(minShare2, 1.0).toDouble //權重，任務數相同，權重高的優先 val taskToWeightRatio1 = runningTasks1.toDouble / s1.weight.toDouble val taskToWeightRatio2 = runningTasks2.toDouble / s2.weight.toDouble var compare: Int = 0 //挨餓的優先 if (s1Needy && !s2Needy) { return true } else if (!s1Needy && s2Needy) { return false } else if (s1Needy && s2Needy) { //都處於挨餓狀態則，需要資源占用比小 的優先 compare = minShareRatio1.compareTo(minShareRatio2)    } else { //都不挨餓，則比較權重比，比例低的優先 compare = taskToWeightRatio1.compareTo(taskToWeightRatio2)   } if (compare < 0) { true } else if (compare > 0) { false } else { //如果都一樣，那麼比較名字,按照字母順序比較，不考慮長度，所以名字比較重要 s1.name < s2.name  } }} 

注：

1. 公平原則本著的原則就是誰最需要就給誰，所以挨餓者優先；

2. 資源占用比這塊有點費解，如果把他理解成一個貪心問題就容易理解了。對於都是出於挨餓狀態的任務可以這樣理解，負載大的即時給你資源你也不一定能有效緩解，莫不如給負載小的，讓其快速使用，完成後可以釋放更多的資源，這是一種貪心策略。如JobA和JobB的Task數量相同都是10，A的minShare是2，B的是5，那占用比為5和2，顯然B的占用比更小，貪心的策略應該給B先調度處理；

3. 對於都處於滿足狀態的，當然誰的權重有著更好的決定性，權重比低得優先（偏向權利大的）；

4. 如果所有上述的比較都相同，那麼名字字典排序靠前的優先（哈哈，名字很重要哦）；名字aaa要比abc優先，所以這裡在給Pool或者TaskSetManager起名字的時候要考慮這一點。

這兩種調度的排序算法針對的可比較對象都是Schedule的具體對象，其(trait可理解成java中接口)定義如下：

private[spark] trait Schedulable {    //指明父對象，即這個Pool或TaskSetManager所屬的調度對象，調度是層級的，是樹狀的    var parent: Pool    // 他擁有的調度對象，即負責管理的調度對象 def schedulableQueue: ConcurrentLinkedQueue[Schedulable]    //負責管理的對象間的排序模型，目前只有FIFO和FAIR兩種算法 def schedulingMode: SchedulingMode    //權重，指的是和同級的相比的權重，權重越大獲得的資源越多 def weight: Int    //最小共享值，指的是可運行需要的最小資源數，即CPU數量 def minShare: Int def runningTasks: Int    //優先級，指的是在同級別中的優先級，優先級高的優先調度 def priority: Int    //這個stageId是對TaskSetManager而言，因為一個Stage的Tasks，實際以一個TaskSet提交 def stageId: Int def name: String def addSchedulable(schedulable: Schedulable): Unit def removeSchedulable(schedulable: Schedulable): Unit def getSchedulableByName(name: String): Schedulable def executorLost(executorId: String, host: String): Unit def checkSpeculatableTasks(): Boolean def getSortedTaskSetQueue: ArrayBuffer[TaskSetManager]} 

目前Spark中有兩種可調度的實體，Pool和TaskSetManager。Pool是一個調度池，Pool裡面還可以有子Pool，Spark中的rootPool即根節點默認是一個無名的Pool。