1、? 資源分配

???????? 通過SparkSubmit進行提交應用后，首先會創建Client將應用程序(字節碼文件.class)包裝成Driver,并將其注冊到Master。Master收到Client的注冊請求后將其加入待調度隊列waitingDrivers，并等待分配執行資源。

1．1 Dirver調度(分配Driver執行容器，1個)

???????? Master中調度程序執行時會為Driver分配一滿足其執行要求的Worker, 并通知Worker啟動將Driver。Worker接到執行Driver指令后創建DriverRunner執行Driver(應用程序mainClass，mainClass執行時其會創建Spark執行上下文環境：SparkContext。伴隨SparkContext會創建DAGScheduler和TaskScheduler分別用于Stage調度和任務調度,并會觸發RDD的Action算子提交job)。

1.2 APP調度(分配Executor， 多個)

???????? 若想Job運行就需要得到執行資源，Dirver成功執行后，會通過SparkDeployScheduler-Backend創建AppClient(包裝App信息，包含可以創建CoarseGrainedExecutorBackend實例Command)，用于向Master匯報資源需求。Master接到AppClient的匯報后，將其加入waittingApps隊列，等待調度。

App調度時會為app分配滿足條件的資源-----Worker（State是Alive，其上并沒有該Application的executor，可用內存滿足要求(spark.executor.memory指定，默認512), 核滿足要求(spark.cores.max, 最大可用core數，若未指定，則為全部資源)），然后通知Woker啟動Excutor. 及向AppClient發送ExecutorAdded消息。

進行調度時，調度程序會根據配制SpreadOutApps = spark.deploy.spreadOut情況決定資源分配方式，若

Master資源調度分配算法：

1.Application的調度算法有兩種，一種是spreadOutApps，另一種是非spreadOutApps。

2.spreadOutApps，會將每個Application要啟動的executor都平均分配到各個worker上去。（比如有10個worker，20個cpu core要分配，那么實際會循環兩遍worker，每個worker分配一個core，最后每個worker分配了2個core，這里的executor數量可能會與spark-submit設置的不一致）

3.非spreadOutApps，將每個Application盡可能分配到盡量少的worker上去。（比如總共有10個worker，每個有10個core，app總共要分配20個core，其實只會分配到兩個worker上，每個worker占滿10個core，其余app只能分配到下一個worker，這里的executor數量可能會與spark-submit設置的不一致）

由分配過程可知， 分配的Excutor個數與CPU核心數有關。當指定完Worker節點后，會在Worker節點創建ExecutorRunner，并啟動，執行App中的Command 去創建并啟動CoarseGrainedExecutorBackend。CoarseGrainedExecutorBackend啟動后，會首先通過傳入的driverUrl這個參數向在CoarseGrainedSchedulerBackend::DriverActor(用于與Master通信，及調度任務)發送RegisterExecutor(executorId, hostPort, cores)，DriverActor會創建executorData（executor信息）加入executorDataMap供后續task使用，并回復RegisteredExecutor，此時CoarseGrainedExecutorBackend會創建一個org.apache.spark.executor.Executor。至此，Executor創建完畢。Executor是直接用于task執行， 是集群中的直接勞動者。

???????? 至此，資源分配結束。當分配完資源后，就可以為依本地性為任務分配具體的執行資源。

????????

2、Stage劃分

???????? 當執行mainClass時，執行到RDD的action算子時，會觸發執行作業(sc.runJob)，最終通過調用DAGScheduler的runJob方法根據RDD信息及action算子要做的操作創建ResultStage(FinalStage)及ActiveJob。

???????? 若ResultStage創建成功的話，根據配制信息及RDD特征可分為本地執行，集群執行。

若“spark.localExecution.enable”指定允許本地運行（默認為：false，不允許），具RDD的action算了允許本地運行allowLocal=true,且RDD只有一個partition的話可以直接以本地線程執行job,無需劃分stage。否則要將job分成多個Stage提交到集群去執行（通過提交ResultStage進行）。

???????? 因為ResultStage提交時，首先會去判斷其是否存在缺失的ParentStage（也就是說是否存在未完成的父Stage）。若有，則其需要等待其父Stage執行完成，才能進行提交執行。
?????? 判斷是否存在Stage的標準是看是否存在ShuffeDependency（Stage的分界線）。提交ResultStage時會根據其finalRDD?的依賴遞歸的尋找其DAG圖中是否存在ShuffeDependency,?若存在，則創建ShuffleMapStage做為finalStage的父Stage以此類似。但至此，只能說存在父Stage并不能說存在缺失的父Stage.?判斷缺失的標準是看其結果成功的輸出信息（status）個數與其處理的分區個數是否相同，如若相同，則說明父Stage已經執行完成,?不存在missing；否則，說明還未完成，存在missing. ?因為將ShuffleMapStage劃分成maptask時，每個Partition對應一個maptask,?每個task會得到一個status輸出結果信息，并在執行結束時將輸出結果上報mapOutputTracker，并更新shuffleStage狀態(將status增加進行其outputLocs列表，并將numAvailableOutputs加1)，若numAvailableOutputs?與?Stage所要處理的partitions一致，說明所有的task都已經執行完成，即Stage執行完成；否則，說明還有task未完成，即Stage未完成。
?????? 由上述分析可知，存在依賴關系的兩個Stage，如果父Stage未執行完成，子Stage不能提交，也就是不能轉變為Taskset加入任務調度隊列。因此其先后順序是嚴格控制的。我們知道只有存在ShuffleDependency時，才會劃分Stage，這也就是說兩個Stage之間是要做Shuffle操作的。根據上述分析可知Shuffle時ShuffleWrite做不完，ShuffleRead不能進行.

3． Task調度

???????? 當Stage不存在缺失的ParentStage時，會將其轉換為TaskSet并提交。轉換時依Stage類型進行轉換：將ResultStage轉換成ResultTask, ShuffleMapStage轉換成ShuffleMapTask. Task個數由Stage中finalRDD 的分區數決定。

???????? 當轉換成的TaskSet提交之后，將其通過taskScheduler包裝成TaskSetManager并添加至調度隊列中（Pool）,等待調度。在包裝成TaskSetManager時，根據task的preferredLocatitions將任務分類存放在pendingTasksForExecutor, pendingTaskForHost, pendingTasksForRack, pendingTaskWithNoPrefs及allPendingTasks中， 前三個列表是是包含關系（本地性越來越低），范圍起來越大，例如：在pendingTasksForExecutor也在pendingTaskForHost，pendingTasksForRack中， 分類的目的是在調度時，依次由本地性高à低的查找task。

???????? 在進行Task調度時，首先根據調度策略將可調度所有taskset進行排序，然后對排好序的taskset待調度列表中的taskset，按序進行分配Executor。再分配Executor時，然后逐個為Executor列表中可用的Executor在此次選擇的taskset中按本地性由高到低查找適配任務。此處任務調度為延遲調度，即若本次調度時間距上一任務結束時間小于當前本地性配制時間則等待，若過了配制時間，本地性要求逐漸降低，再去查找適配的task。當選定某一task后后將其加入runningtask列表，當其執行完成時會加入success列表，下次調度時就會過濾過存在這兩個列表中的任務，避免重復調度。

???????? 當一個任務執行結束時，會將其從runningtask中移除，并加入success,并會適放其占用的執行資源,供后序task使用， 將判斷其執行成功的task數與此taskset任務總數相等時，意為taskset中所有任務執行結束，也就是taskset結束。此時會將taskset移除出可調度隊列。

???????? 重復上述過程直到taskset待調度列表為空。即所有作業（job）執行完成。

Spark中的調度模式主要有兩種：FIFO和FAIR。默認情況下Spark的調度模式是FIFO（先進先出），誰先提交誰先執行，后面的任務需要等待前面的任務執行。而FAIR（公平調度）模式支持在調度池中為任務進行分組，不同的調度池權重不同，任務可以按照權重來決定執行順序。對這兩種調度模式的具體實現，接下來會根據spark-1.6.0的源碼來進行詳細的分析。使用哪種調度器由參數spark.scheduler.mode來設置，可選的參數有FAIR和FIFO，默認是FIFO

?

總結

以上是生活随笔為你收集整理的Spark 资源调度及任务调度的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。