通過文章“Spark核心概念RDD”我們知道竭讳，Spark的核心是根據(jù)RDD來實現(xiàn)的，Spark Scheduler則為Spark核心實現(xiàn)的重要一環(huán)黔帕，其作用就是任務調度代咸。Spark的任務調度就是如何組織任務去處理RDD中每個分區(qū)的數(shù)據(jù)，根據(jù)RDD的依賴關系構建DAG成黄，基于DAG劃分Stage呐芥，將每個Stage中的任務發(fā)到指定節(jié)點運行》芩辏基于Spark的任務調度原理思瘟，我們可以合理規(guī)劃資源利用，做到盡可能用最少的資源高效地完成任務計算闻伶。

分布式運行框架

Spark可以部署在多種資源管理平臺滨攻，例如Yarn、Mesos等，Spark本身也實現(xiàn)了一個簡易的資源管理機制光绕，稱之為Standalone模式女嘲。由于工作中接觸較多的是Saprk on Yarn，不做特別說明诞帐，以下所述均表示Spark-on-Yarn欣尼。Spark部署在Yarn上有兩種運行模式，分別為yarn-client和yarn-cluster模式停蕉，它們的區(qū)別僅僅在于Spark Driver是運行在Client端還是ApplicationMater端愕鼓。如下圖所示為Spark部署在Yarn上，以yarn-cluster模式運行的分布式計算框架慧起。

其中藍色部分是Spark里的概念菇晃，包括Client、ApplicationMaster蚓挤、Driver和Executor磺送，其中Client和ApplicationMaster主要是負責與Yarn進行交互；Driver作為Spark應用程序的總控屈尼，負責分發(fā)任務以及監(jiān)控任務運行狀態(tài)册着；Executor負責執(zhí)行任務，并上報狀態(tài)信息給Driver脾歧，從邏輯上來看Executor是進程甲捏，運行在其中的任務是線程，所以說Spark的任務是線程級別的鞭执。通過下面的時序圖可以更清晰地理解一個Spark應用程序從提交到運行的完整流程司顿。

提交一個Spark應用程序，首先通過Client向ResourceManager請求啟動一個Application兄纺，同時檢查是否有足夠的資源滿足Application的需求大溜，如果資源條件滿足，則準備ApplicationMaster的啟動上下文估脆，交給ResourceManager钦奋，并循環(huán)監(jiān)控Application狀態(tài)。

當提交的資源隊列中有資源時疙赠，ResourceManager會在某個NodeManager上啟動ApplicationMaster進程付材，ApplicationMaster會單獨啟動Driver后臺線程，當Driver啟動后圃阳，ApplicationMaster會通過本地的RPC連接Driver厌衔，并開始向ResourceManager申請Container資源運行Executor進程（一個Executor對應與一個Container），當ResourceManager返回Container資源捍岳，則在對應的Container上啟動Executor富寿。

Driver線程主要是初始化SparkContext對象睬隶，準備運行所需的上下文，然后一方面保持與ApplicationMaster的RPC連接页徐，通過ApplicationMaster申請資源苏潜，另一方面根據(jù)用戶業(yè)務邏輯開始調度任務，將任務下發(fā)到已有的空閑Executor上变勇。

當ResourceManager向ApplicationMaster返回Container資源時窖贤，ApplicationMaster就嘗試在對應的Container上啟動Executor進程，Executor進程起來后贰锁，會向Driver注冊，注冊成功后保持與Driver的心跳滤蝠，同時等待Driver分發(fā)任務豌熄，當分發(fā)的任務執(zhí)行完畢后，將任務狀態(tài)上報給Driver物咳。

Driver把資源申請的邏輯給抽象出來锣险，以適配不同的資源管理系統(tǒng)，所以才間接地通過ApplicationMaster去和Yarn打交道览闰。

從上述時序圖可知芯肤，Client只管提交Application并監(jiān)控Application的狀態(tài)。對于Spark的任務調度主要是集中在兩個方面: 資源申請和任務分發(fā)压鉴，其主要是通過ApplicationMaster崖咨、Driver以及Executor之間來完成，下面詳細剖析Spark任務調度每個細節(jié)油吭。

Spark任務調度總覽

當Driver起來后击蹲，Driver則會根據(jù)用戶程序邏輯準備任務，并根據(jù)Executor資源情況逐步分發(fā)任務婉宰。在詳細闡述任務調度前歌豺，首先說明下Spark里的幾個概念。一個Spark應用程序包括Job心包、Stage以及Task三個概念：

• Job是以Action方法為界类咧，遇到一個Action方法則觸發(fā)一個Job；
• Stage是Job的子集蟹腾，以RDD寬依賴(即Shuffle)為界痕惋，遇到Shuffle做一次劃分；
• Task是Stage的子集岭佳，以并行度(分區(qū)數(shù))來衡量血巍，分區(qū)數(shù)是多少，則有多少個task珊随。

Spark的任務調度總體來說分兩路進行述寡，一路是Stage級的調度柿隙，一路是Task級的調度，總體調度流程如下圖所示鲫凶。

Spark RDD通過其Transactions操作禀崖，形成了RDD血緣關系圖，即DAG螟炫，最后通過Action的調用波附，觸發(fā)Job并調度執(zhí)行。DAGScheduler負責Stage級的調度昼钻，主要是將DAG切分成若干Stages掸屡，并將每個Stage打包成TaskSet交給TaskScheduler調度。TaskScheduler負責Task級的調度然评，將DAGScheduler給過來的TaskSet按照指定的調度策略分發(fā)到Executor上執(zhí)行仅财，調度過程中SchedulerBackend負責提供可用資源，其中SchedulerBackend有多種實現(xiàn)碗淌，分別對接不同的資源管理系統(tǒng)盏求。有了上述感性的認識后，下面這張圖描述了Spark-On-Yarn模式下在任務調度期間亿眠，ApplicationMaster碎罚、Driver以及Executor內部模塊的交互過程。

Driver初始化SparkContext過程中纳像，會分別初始化DAGScheduler荆烈、TaskScheduler、SchedulerBackend以及HeartbeatReceiver爹耗，并啟動SchedulerBackend以及HeartbeatReceiver耙考。SchedulerBackend通過ApplicationMaster申請資源，并不斷從TaskScheduler中拿到合適的Task分發(fā)到Executor執(zhí)行潭兽。HeartbeatReceiver負責接收Executor的心跳信息倦始，監(jiān)控Executor的存活狀況，并通知到TaskScheduler山卦。下面著重剖析DAGScheduler負責的Stage調度以及TaskScheduler負責的Task調度鞋邑。

Stage級的調度

Spark的任務調度是從DAG切割開始，主要是由DAGScheduler來完成账蓉。當遇到一個Action操作后就會觸發(fā)一個Job的計算枚碗，并交給DAGScheduler來提交，下圖是涉及到Job提交的相關方法調用流程圖铸本。

Job由最終的RDD和Action方法封裝而成肮雨，SparkContext將Job交給DAGScheduler提交，它會根據(jù)RDD的血緣關系構成的DAG進行切分箱玷，將一個Job劃分為若干Stages怨规，具體劃分策略是陌宿，由最終的RDD不斷通過依賴回溯判斷父依賴是否是款依賴，即以Shuffle為界波丰，劃分Stage壳坪，窄依賴的RDD之間被劃分到同一個Stage中，可以進行pipeline式的計算掰烟，如上圖紫色流程部分爽蝴。劃分的Stages分兩類，一類叫做ResultStage纫骑，為DAG最下游的Stage蝎亚，由Action方法決定，另一類叫做ShuffleMapStage先馆，為下游Stage準備數(shù)據(jù)颖对，下面看一個簡單的例子WordCount。

Job由saveAsTextFile觸發(fā)磨隘，該Job由RDD-3和saveAsTextFile方法組成，根據(jù)RDD之間的依賴關系從RDD-3開始回溯搜索顾患，直到沒有依賴的RDD-0番捂，在回溯搜索過程中，RDD-3依賴RDD-2江解，并且是寬依賴设预，所以在RDD-2和RDD-3之間劃分Stage，RDD-3被劃到最后一個Stage犁河，即ResultStage中鳖枕，RDD-2依賴RDD-1，RDD-1依賴RDD-0桨螺，這些依賴都是窄依賴宾符，所以將RDD-0、RDD-1和RDD-2劃分到同一個Stage灭翔，即ShuffleMapStage中魏烫，實際執(zhí)行的時候，數(shù)據(jù)記錄會一氣呵成地執(zhí)行RDD-0到RDD-2的轉化肝箱。不難看出哄褒，其本質上是一個深度優(yōu)先搜索算法。

一個Stage是否被提交煌张，需要判斷它的父Stage是否執(zhí)行呐赡，只有在父Stage執(zhí)行完畢才能提交當前Stage，如果一個Stage沒有父Stage骏融，那么從該Stage開始提交链嘀。Stage提交時會將Task信息（分區(qū)信息以及方法等）序列化并被打包成TaskSet交給TaskScheduler萌狂，一個Partition對應一個Task，另一方面監(jiān)控Stage的運行狀態(tài)管闷，只有Executor丟失或者Task由于Fetch失敗才需要重新提交失敗的Stage以調度運行失敗的任務粥脚，其他類型的Task失敗會在TaskScheduler的調度過程中重試。

相對來說DAGScheduler做的事情較為簡單包个，僅僅是在Stage層面上劃分DAG刷允，提交Stage并監(jiān)控相關狀態(tài)信息。TaskScheduler則相對較為復雜碧囊，下面詳細闡述其細節(jié)树灶。

Task級的調度

Spark Task的調度是由TaskScheduler來完成，由前文可知糯而，DAGScheduler將Stage打包到TaskSet交給TaskScheduler天通，TaskScheduler會將其封裝為TaskSetManager加入到調度隊列中，TaskSetManager負責監(jiān)控管理同一個Stage中的Tasks熄驼，TaskScheduler就是以TaskSetManager為單元來調度任務像寒。前面也提到，TaskScheduler初始化后會啟動SchedulerBackend瓜贾，它負責跟外界打交道诺祸，接收Executor的注冊信息，并維護Executor的狀態(tài)祭芦，所以說SchedulerBackend是管“糧食”的筷笨，同時它在啟動后會定期地去“詢問”TaskScheduler有沒有任務要運行，也就是說龟劲，它會定期地“問”TaskScheduler“我有這么余量胃夏，你要不要啊”，TaskScheduler在SchedulerBackend“問”它的時候昌跌，會從調度隊列中按照指定的調度策略選擇TaskSetManager去調度運行仰禀，大致方法調用流程如下圖所示。

調度策略

前面講到蚕愤，TaskScheduler會先把DAGScheduler給過來的TaskSet封裝成TaskSetManager扔到任務隊列里悼瘾，然后再從任務隊列里按照一定的規(guī)則把它們取出來在SchedulerBackend給過來的Executor上運行。這個調度過程實際上還是比較粗粒度的审胸，是面向TaskSetManager的亥宿。

TaskScheduler是以樹的方式來管理任務隊列，樹中的節(jié)點類型為Schdulable砂沛，葉子節(jié)點為TaskSetManager烫扼，非葉子節(jié)點為Pool，下圖是它們之間的繼承關系碍庵。

TaskScheduler支持兩種調度策略映企，一種是FIFO悟狱，也是默認的調度策略，另一種是FAIR堰氓。在TaskScheduler初始化過程中會實例化rootPool挤渐，表示樹的根節(jié)點，是Pool類型双絮。如果是采用FIFO調度策略浴麻，則直接簡單地將TaskSetManager按照先來先到的方式入隊，出隊時直接拿出最先進隊的TaskSetManager囤攀，其樹結構大致如下圖所示软免，TaskSetManager保存在一個FIFO隊列中。

在闡述FAIR調度策略前焚挠，先貼一段使用FAIR調度策略的應用程序代碼膏萧，后面針對該代碼邏輯來詳細闡述FAIR調度的實現(xiàn)細節(jié)。

object MultiJobTest {
  // spark.scheduler.mode=FAIR
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().getOrCreate()

    val rdd = spark.sparkContext.textFile(...)
      .map(_.split("\\s+"))
      .map(x => (x(0), x(1)))

    val jobExecutor = Executors.newFixedThreadPool(2)

    jobExecutor.execute(new Runnable {
      override def run(): Unit = {
        spark.sparkContext.setLocalProperty("spark.scheduler.pool", "count-pool")
        val cnt = rdd.groupByKey().count()
        println(s"Count: $cnt")
      }
    })

    jobExecutor.execute(new Runnable {
      override def run(): Unit = {
        spark.sparkContext.setLocalProperty("spark.scheduler.pool", "take-pool")
        val data = rdd.sortByKey().take(10)
        println(s"Data Samples: ")
        data.foreach { x => println(x.mkString(", ")) }
      }
    })

    jobExecutor.shutdown()
    while (!jobExecutor.isTerminated) {}
    println("Done!")
  }
}

上述應用程序中使用兩個線程分別調用了Action方法蝌衔，即有兩個Job會并發(fā)提交榛泛，但是不管怎樣，這兩個Job被切分成若干TaskSet后終究會被交到TaskScheduler這里統(tǒng)一管理噩斟，其調度樹大致如下圖所示挟鸠。

    var sortedTaskSetQueue = new ArrayBuffer[TaskSetManager]
    val sortedSchedulableQueue = schedulableQueue.asScala.toSeq.sortWith(taskSetSchedulingAlgorithm.comparator)
    for (schedulable <- sortedSchedulableQueue) {
      sortedTaskSetQueue ++= schedulable.getSortedTaskSetQueue
    }
    sortedTaskSetQueue
  }

使用FAIR調度策略時佳遂，上面代碼中的taskSetSchedulingAlgorithm的類型為FairSchedulingAlgorithm营袜，排序過程的比較是基于Fair-share來比較的，每個要排序的對象包含三個屬性: runningTasks值（正在運行的Task數(shù)）丑罪、minShare值荚板、weight值，比較時會綜合考量runningTasks值吩屹，minShare以及weight值跪另。如果A對象的runningTasks大于它的minShare，B對象的runningTasks小于它的minShare煤搜，那么B排在A前面免绿；如果A、B對象的runningTasks都小于它們的minShare擦盾，那么就比較runningTasks與minShare的比值嘲驾，誰小誰排前面淌哟；如果A、B對象的runningTasks都大于它們的minShare辽故，那么就比較runningTasks與weight的比值徒仓，誰小誰排前面。整體上來說就是通過minShare和weight這兩個參數(shù)控制比較過程誊垢，可以做到不讓資源被某些長時間Task給一直占了掉弛。

從調度隊列中拿到TaskSetManager后，那么接下來的工作就是TaskSetManager按照一定的規(guī)則一個個取出Task給TaskScheduler彤枢，TaskScheduler再交給SchedulerBackend去發(fā)到Executor上執(zhí)行狰晚。前面也提到，TaskSetManager封裝了一個Stage的所有Task缴啡，并負責管理調度這些Task壁晒。

本地化調度

從調度隊列中拿到TaskSetManager后，那么接下來的工作就是TaskSetManager按照一定的規(guī)則一個個取出Task給TaskScheduler业栅，TaskScheduler再交給SchedulerBackend去發(fā)到Executor上執(zhí)行玩荠。前面也提到伴逸，TaskSetManager封裝了一個Stage的所有Task，并負責管理調度這些Task。

在TaskSetManager初始化過程中箕昭，會對Tasks按照Locality級別進行分類，Task的Locality有五種克胳，優(yōu)先級由高到低順序：PROCESS_LOCAL(指定的Executor)类缤，NODE_LOCAL(指定的主機節(jié)點)，NO_PREF(無所謂)文兢，RACK_LOCAL(指定的機架)晤斩，ANY(滿足不了Task的Locality就隨便調度)。這五種Locality級別存在包含關系姆坚，RACK_LOCAL包含NODE_LOCAL澳泵，NODE_LOCAL包含PROCESS_LOCAL，然而ANY包含其他所有四種兼呵。初始化階段在對Task分類時兔辅，根據(jù)Task的preferredLocations判斷它屬于哪個Locality級別，屬于PROCESS_LOCAL的Task同時也會被加入到NODE_LOCAL击喂、RACK_LOCAL類別中维苔，比如，一個Task的preferredLocations指定了在Executor-2上執(zhí)行懂昂，那么它屬于Executor-2對應的PROCESS_LOCAL類別蕉鸳，同時也把他加入到Executor-2所在的主機對應的NODE_LOCAL類別，Executor-2所在的主機的機架對應的RACK_LOCAL類別中，以及ANY類別潮尝，這樣在調度執(zhí)行時榕吼，滿足不了PROCESS_LOCAL，就逐步退化到NODE_LOCAL勉失，RACK_LOCAL羹蚣，ANY。

TaskSetManager在決定調度哪些Task時乱凿，是通過上面流程圖中的resourceOffer方法來實現(xiàn)顽素，為了盡可能地將Task調度到它的preferredLocations上，它采用一種延遲調度算法徒蟆。resourceOffer方法原型如下胁出，參數(shù)包括要調度任務的Executor Id、主機地址以及最大可容忍的Locality級別段审。

def resourceOffer(
      execId: String,
      host: String,
      maxLocality: TaskLocality.TaskLocality)
    : Option[TaskDescription]

延遲調度算法的大致流程如下圖所示全蝶。

首先看是否存在execId對應的PROCESS_LOCAL類別的任務，如果存在寺枉，取出來調度抑淫，否則根據(jù)當前時間，判斷是否超過了PROCESS_LOCAL類別最大容忍的延遲姥闪，如果超過始苇，則退化到下一個級別NODE_LOCAL，否則等待不調度筐喳。退化到下一個級別NODE_LOCAL后調度流程也類似催式，看是否存在host對應的NODE_LOCAL類別的任務，如果存在避归，取出來調度荣月，否則根據(jù)當前時間，判斷是否超過了NODE_LOCAL類別最大容忍的延遲槐脏，如果超過，則退化到下一個級別RACK_LOCAL撇寞，否則等待不調度顿天，以此類推…..。當不滿足Locatity類別會選擇等待蔑担，直到下一輪調度重復上述流程牌废，如果你比較激進，可以調大每個類別的最大容忍延遲時間啤握，如果不滿足Locatity時就會等待多個調度周期鸟缕，直到滿足或者超過延遲時間退化到下一個級別為止。

失敗重試與黑名單機制

除了選擇合適的Task調度運行外，還需要監(jiān)控Task的執(zhí)行狀態(tài)懂从，前面也提到授段，與外部打交道的是SchedulerBackend，Task被提交到Executor啟動執(zhí)行后番甩，Executor會將執(zhí)行狀態(tài)上報給SchedulerBackend侵贵，SchedulerBackend則告訴TaskScheduler，TaskScheduler找到該Task對應的TaskSetManager缘薛，并通知到該TaskSetManager窍育，這樣TaskSetManager就知道Task的失敗與成功狀態(tài)，對于失敗的Task宴胧，會記錄它失敗的次數(shù)漱抓，如果失敗次數(shù)還沒有超過最大重試次數(shù)，那么就把它放回待調度的Task池子中恕齐，否則整個Application失敗乞娄。

在記錄Task失敗次數(shù)過程中，會記錄它上一次失敗所在的Executor Id和Host檐迟，這樣下次再調度這個Task時补胚，會使用黑名單機制，避免它被調度到上一次失敗的節(jié)點上追迟，起到一定的容錯作用溶其。黑名單記錄Task上一次失敗所在的Executor Id和Host，以及其對應的“黑暗”時間敦间，“黑暗”時間是指這段時間內不要再往這個節(jié)點上調度這個Task了瓶逃。

推測式執(zhí)行

TaskScheduler在啟動SchedulerBackend后，還會啟動一個后臺線程專門負責推測任務的調度廓块，推測任務是指對一個Task在不同的Executor上啟動多個實例厢绝，如果有Task實例運行成功，則會干掉其他Executor上運行的實例带猴。推測調度線程會每隔固定時間檢查是否有Task需要推測執(zhí)行昔汉，如果有，則會調用SchedulerBackend的reviveOffers去嘗試拿資源運行推測任務拴清。

檢查是否有Task需要推測執(zhí)行的邏輯最后會交到TaskSetManager靶病，TaskSetManager采用基于統(tǒng)計的算法，檢查Task是否需要推測執(zhí)行口予，算法流程大致如下圖所示娄周。

TaskSetManager首先會統(tǒng)計成功的Task數(shù)，當成功的Task數(shù)超過75%(可通過參數(shù)spark.speculation.quantile控制)時沪停，再統(tǒng)計所有成功的Tasks的運行時間煤辨，得到一個中位數(shù)裳涛，用這個中位數(shù)乘以1.5(可通過參數(shù)spark.speculation.multiplier控制)得到運行時間門限，如果在運行的Tasks的運行時間超過這個門限众辨，則對它啟用推測端三。算法邏輯較為簡單，其實就是對那些拖慢整體進度的Tasks啟用推測泻轰，以加速整個TaskSet即Stage的運行技肩。

資源申請機制

在前文已經提過，ApplicationMaster和SchedulerBackend起來后浮声，SchedulerBackend通過ApplicationMaster申請資源虚婿，ApplicationMaster就是用來專門適配YARN申請Container資源的，當申請到Container泳挥，會在相應Container上啟動Executor進程然痊，其他事情就交給SchedulerBackend。Spark早期版本只支持靜態(tài)資源申請屉符，即一開始就指定用多少資源剧浸，在整個Spark應用程序運行過程中資源都不能改變，后來支持動態(tài)Executor申請矗钟，用戶不需要指定確切的Executor數(shù)量唆香，Spark會動態(tài)調整Executor的數(shù)量以達到資源利用的最大化。

靜態(tài)資源申請

靜態(tài)資源申請是用戶在提交Spark應用程序時吨艇，就要提前估計應用程序需要使用的資源躬它，包括Executor數(shù)(num_executor)、每個Executor上的core數(shù)(executor_cores)东涡、每個Executor的內存(executor_memory)以及Driver的內存(driver_memory)冯吓。

在估計資源使用時，應當首先了解這些資源是怎么用的疮跑。任務的并行度由分區(qū)數(shù)(Partitions)決定组贺，一個Stage有多少分區(qū)，就會有多少Task祖娘。每個Task默認占用一個Core失尖，一個Executor上的所有core共享Executor上的內存，一次并行運行的Task數(shù)等于num_executor*executor_cores渐苏，如果分區(qū)數(shù)超過該值掀潮，則需要運行多個輪次，一般來說建議運行3～5輪較為合適整以，否則考慮增加num_executor或executor_cores胧辽。由于一個Executor的所有tasks會共享內存executor_memory峻仇，所以建議executor_cores不宜過大公黑。executor_memory的設置則需要綜合每個分區(qū)的數(shù)據(jù)量以及是否有緩存等邏輯。下圖描繪了一個應用程序內部資源利用情況。

動態(tài)資源申請

動態(tài)資源申請目前只支持到Executor凡蚜，即可以不用指定num_executor人断，通過參數(shù)spark.dynamicAllocation.enabled來控制。由于許多Spark應用程序一開始可能不需要那么多Executor或者其本身就不需要太多Executor朝蜘，所以不必一次性申請那么多Executor恶迈，根據(jù)具體的任務數(shù)動態(tài)調整Executor的數(shù)量，盡可能做到資源的不浪費谱醇。由于動態(tài)Executor的調整會導致Executor動態(tài)的添加與刪除暇仲，如果刪除Executor，其上面的中間Shuffle結果可能會丟失副渴，這就需要借助第三方的ShuffleService了奈附，如果Spark是部署在Yarn上，則可以在Yarn上配置Spark的ShuffleService煮剧，具體操作僅需做兩點:

1.首先在yarn-site.xml中加上如下配置：

<property>
  <name>yarn.nodemanager.aux-services</name>
  <value>mapreduce_shuffle,spark_shuffle</value>
</property>
<property>
  <name>yarn.nodemanager.aux-services.spark_shuffle.class</name>
  <value>org.apache.spark.network.yarn.YarnShuffleService</value>
</property>
<property>
  <name>spark.shuffle.service.port</name>
  <value>7337</value>
</property>

2. 將Spark ShuffleService jar包$SPARK_HOME/lib/spark-*-yarn-shuffle.jar拷貝到每臺NodeManager的$HADOOP_HOME/share/hadoop/yarn/lib/下斥滤，并重啟所有的NodeManager。

當啟用動態(tài)Executor申請時勉盅，在SparkContext初始化過程中會實例化ExecutorAllocationManager佑颇，它是被用來專門控制動態(tài)Executor申請邏輯的，動態(tài)Executor申請是一種基于當前Task負載壓力實現(xiàn)動態(tài)增刪Executor的機制草娜。一開始會按照參數(shù)spark.dynamicAllocation.initialExecutors設置的初始Executor數(shù)申請挑胸，然后根據(jù)當前積壓的Task數(shù)量，逐步增長申請的Executor數(shù)驱还，如果當前有積壓的Task嗜暴，那么取積壓的Task數(shù)和spark.dynamicAllocation.maxExecutors中的最小值作為Executor數(shù)上限，每次新增加申請的Executor為2的次方议蟆，即第一次增加1闷沥，第二次增加2，第三次增加4咐容，…舆逃。另一方面，如果一個Executor在一段時間內都沒有Task運行戳粒，則將其回收路狮，但是在Remove Executor時，要保證最少的Executor數(shù)蔚约，該值通過參數(shù)spark.dynamicAllocation.minExecutors來控制奄妨，如果Executor上有Cache的數(shù)據(jù)，則永遠不會被Remove苹祟，以保證中間數(shù)據(jù)不丟失砸抛。

結語

本文詳細闡述了Spark的任務調度评雌，著重討論Spark on Yarn的部署調度，剖析了從應用程序提交到運行的全過程直焙。Spark Schedule算是Spark中的一個大模塊景东，它負責任務下發(fā)與監(jiān)控等，基本上扮演了Spark大腦的角色奔誓。了解Spark Schedule有助于幫助我們清楚地認識Spark應用程序的運行軌跡斤吐，同時在我們實現(xiàn)其他系統(tǒng)時，也可以借鑒Spark的實現(xiàn)厨喂。

轉載：http://sharkdtu.com/posts/spark-scheduler.html