Spark Core解析 2：Scheduler 調度體系

時間 2020-02-13

標籤 spark core 解析 scheduler 調度體系欄目 Spark 简体版

原文原文鏈接

Spark Core解析 2：Scheduler 調度體系

Overview

調度系統，是貫穿整個Spark應用的主心骨，從調度系統開始入手瞭解Spark Core，比較容易理清頭緒。html

Spark的資源調度採用的是常見的兩層調度，底層資源的管理和分配是第一層調度，交給YARN、Mesos或者Spark的Standalone集羣處理，Application從第一層調度拿到資源後，還要進行內部的任務和資源調度，將任務和資源進行匹配，這是第二層調度，本文講的就是這第二層調度。算法

Spark的調度體系涉及的任務包括3個粒度，分別是Job、Stage、Task。
Job表明用戶提交的一系列操做的整體，一個具體的計算任務，有明確的輸入輸出，一個Job由多個Stage組成；
一個Stage表明Job計算流程的一個組成部分，一個階段，包含多個Task；
一個Task表明對一個分區的數據進行計算的具體任務。shell

層級關係：Job > Stage > Taskapache

在Spark Core 解析：RDD 彈性分佈式數據集中，已經解釋了RDD之間的依賴，以及如何組成RDD血緣圖。segmentfault

因此本文主要目的就是解釋清楚：Scheduler將RDD血緣圖轉變成Stage DAG，而後生成Task，最後提交給Executor去執行的過程。後端

Stage

Job的不一樣分區的計算一般能夠並行，可是有些計算須要將數據進行從新分區，這個過程稱做shuffle(混洗)。Shuffle的過程是無法徹底並行的，這時候就會出現task之間的等待，task的數量也可能發生變化，因此Spark中以shuffle爲邊界，對task進行劃分，劃分出來的每段稱爲Stage。app

Stage表明一組能夠並行的執行相同計算的task，每一個任務必須有相同的分區規則，這樣一個stage中是沒有shuffle的。異步

在一個Spark App中，stage有一個全局惟一ID，stage id是自增的。async

Stage分爲兩種：分佈式

ResultStage：最後執行的stage，負責Job最終的結果輸出，每一個Job有且僅有一個ResultStage；
ShuffleMapStage：該stage的輸出不是最終結果，而是其餘stage的輸入數據，一般涉及一次shuffle計算。

stage建立流程：

從最終執行action的RDD開始，沿着RDD依賴關係遍歷，

一旦發現某個RDD的dependency是ShuffleDependency，就建立一個ShuffleMapStage。

最後建立ResultStage。

example 1

val rg=sc.parallelize(List((1,10),(2,20)))
rg.reduceByKey(_+_).collect

這裏reduceByKey操做引發了一次shuffle，因此job被切分紅了2個stage。

example 2

val rddA=sc.parallelize(List((1,"a"),(2,"b"),(3,"c")))
val rddB=sc.parallelize(List((1,"A"),(2,"B"),(3,"C")))
rddA.join(rddB).collect

join操做致使rddA和rddB都進行了一次shuffle，因此有3個stage。

example 3

import org.apache.spark.HashPartitioner
val rddA=sc.parallelize(List((1,"a"),(2,"b"),(3,"c"))).partitionBy(new HashPartitioner(3))
val rddB=sc.parallelize(List((1,"A"),(2,"B"),(3,"C")))
rddA.join(rddB).collect

WHAT ?

由於rddA已經定義了Partitioner，這裏join操做會保留rddA的分區方式，因此對rddA的依賴是OneToOneDepenency，而對於rddB則是ShuffleDependency。

探索：一個RDD被依賴屢次，會如何

val rddA=sc.parallelize(List((1,"a"),(2,"b"),(3,"c")))
rddA join rddA collect

一個RDD被兩個stage使用了。

小結

綜上，stage的劃分必定是依據shuffle即ShuffleDependency，跟算子和RDD變量的定義沒有很強的關係，example2和3中的join操做rddA.join(rddB).collect看起來如出一轍，但實際產生的stage劃分卻差異很大。

Task

與stage對應，task也分爲兩種：

ShuffleMapTask：即ShuffleMapStage中的task，主要完成map、shuffle計算。
ResultTask：ResultStage中的task，主要完成最終結果輸出或者返回結果給driver的任務。

一個stage有多少個partition就會建立多少個task，好比一個ShuffleMapStage有10個partition，那麼就會建立10個ShuffleMapTask。

一個Stage中的全部task組成一個TaskSet。

Job Submit

graph TB
R(RDD.action)-->S(SparkContext.runJob)-- RDD -->D(DAGScheduler.runJob)
-- TaskSet -->T(TaskScheduler.submitTasks)-- TaskDescription -->E(Executor.launchTask)

RDD在action操做中經過SparkContext.runJob方法觸發Job執行流程，該方法將調用DagScheduler.runJob方法，將RDD傳入DagScheduler。而後，DAGScheduler建立TaskSet提交給TaskScheduler，TaskScheduler再將TaskSet封裝成TaskDescription發送給Executor，最後Executor會將TaskDescription提交給線程池來運行。

Stage Scheduler(high-level)

DagScheduler

Stage級別的調度是DagScheduler負責的，也是Spark調度體系的核心。

DagScheduler的工做模式

sequenceDiagram
    participant M as main thread
    participant L as eventProcessLoop
    participant E as event thread
    M-->>L: post event
    E-->>L: handle event

DagScheduler內部維護了一個事件消息總線eventProcessLoop(類型爲DAGSchedulerEventProcessLoop)，其實就是一個用來存儲DAGSchedulerEvent類型數據的隊列。

當DagScheduler的一些方法被調用的時候（如submitJob方法），並不會在主線程中處理該任務，而是post一個event(如JobSubmitted)到eventProcessLoop。eventProcessLoop中有一個守護線程，會不斷的依次從隊列中取出event，而後調用對應的handle(如handleJobSubmitted)方法來執行具體的任務。

Stage調度流程

1.submit job
DagScheduler.runJob方法會調用submitJob方法，向eventProcessLoop發送一個JobSubmitted類型的消息，其中包含了RDD等信息。當eventProcessLoop接收到JobSubmitted類型的消息，會調用DagScheduler.handleJobSubmitted方法來處理消息。

sequenceDiagram
    participant M as main thread(runJob)
    participant L as eventProcessLoop
    participant E as event thread(handleJobSubmitted)
    M-->>L: post JobSubmitted event
    E-->>L: handle JobSubmitted event

2.create stage
- DagScheduler在它的handleJobSubmitted方法中開始建立ResultStage。ResultStage中包含了最終執行action的finalRDD，以及計算函數func。
- ResultStage有個parents屬性，這個屬性是個列表，也就是說能夠有多個parent stage。建立ResultStage時須要先建立它的parent stage來填充這個屬性，也就是說要建立ResultStage直接依賴的全部ShuffleMapStage。
- 經過stage.rdd.dependencies屬性，採用寬度優先遍歷，一旦發現某個RDD(假設叫rddA)的dependency是ShuffleDependency，就建立一個ShuffleMapStage，ShuffleMapStage中包含的關鍵信息與ResultStage不一樣，是rddA的ShuffleDependency和rddA的ShuffleDependency.rdd，也就是說新建立的ShuffleMapStage持有的信息是他自身的最後一個RDD和該RDD的子RDD的dependency。
- 建立一個ShuffleMapStage的過程同理會須要建立它的parent stage，也是若干ShuffleMapStage。如此遞歸下去，直到建立完全部的ShuffleMapStage，最後才完成ResultStage的建立。最後建立出來的這些Stage(若干ShuffleMapStage加一個ResultStage)，經過parent屬性串起來，就像這樣
```
graph TD
A[ResultStage]-- parent -->B[ShuffleMapStage 1]
A-- parent -->C[ShuffleMapStage 2]
B-- parent -->D[ShuffleMapStage 3]
```
  這就生成了所謂的DAG圖，可是這個圖的指向跟執行順序是反過來的，若是按執行順序來畫DAG圖，就是常見的形式了：
```
graph TD
D[ShuffleMapStage 3]-->C[ShuffleMapStage 2]
C[ShuffleMapStage 2]-->A[ResultStage]
B[ShuffleMapStage 1]-->A[ResultStage]
```
3.submit stage
DagScheduler.handleJobSubmitted方法建立好ResultStage後會提交這個stage(submitStage方法)，在提交一個stage的時候，會要先提交它的parent stage,也是經過遞歸的形式，直到一個stage的全部parent stage都被提交了，它本身才能被提交，若是一個stage的parent尚未完成，則會把這個stage加入waitingStages。也就是說，DAG圖中前面的stage會被先提交。當一個stage的parent都準備好了，也就是執行完了，它纔會進入submitMissingTasks的環節。
4.submit task
Task是在DagScheduler（不是TaskScheduler）的submitMissingTasks方法中建立的，包括ShuffleMapTask和ResultTask，與Stage對應。歸屬於同一個stage的這批Task組成一個TaskSet集合，最後提交給TaskScheduler的就是這個TaskSet集合。

Task Scheduler(low-level)

Task的調度工做是由TaskScheduler與SchedulerBackend緊密合做，共同完成的。

TaskScheduler是task級別的調度器，主要做用是管理task的調度和提交，是Spark底層的調度器。

SchedulerBackend是TaskScheduler的後端服務，有獨立的線程，全部的Executor都會註冊到SchedulerBackend，主要做用是進行資源分配、將task分配給executor等。

Task調度流程

第一個線程是DAGScheduler的事件處理線程，在其中，Task先通過DAGScheduler（藍色箭頭表示）封裝成TaskSet，再由TaskScheduler（綠色箭頭）封裝成TaskSetManager，並加入調度隊列中。

SchedulerBackend在收到ReviveOffers消息時，會從線程池取一個線程進行makeOffers操做，WorkerOffer建立後傳遞給TaskScheduler進行分配。

圖中第二個線程就是SchedulerBackend的一個事件分發線程，從Pool中取出最優先的TaskSetManager，而後將WorkerOffer與其中的Task進行配對，生成TaskDescription，發送給WorkerOffer指定的Executor去執行。

工做流程

1 DAGScheduler(submitMissingTasks方法中)調用TaskScheduler.submitTasks()建立並提交TaskSet給TaskScheduler；
2 TaskScheduler拿到TaskSet後會建立一個TaskSetManager來管理它，而且把TaskSetManager添加到rootPool調度池中；
3 調用SchedulerBackend.reviveOffers()方法；
4 SchedulerBackend發送ReviveOffers消息給DriverEndpoint；
5 DriverEndpoint收到ReviveOffers消息後，會調用makeOffers()方法建立WorkerOffer，並經過TaskScheduler.resourceOffers()返回offer；
6 TaskScheduler從rootPool獲取按調度算法排序後的TaskSetManager列表，取第一個TaskSetManager，逐個給TaskSet的Task分配WorkerOffer，生成TaskDescription(包含offer信息)；
7 調用SchedulerBackend.DriverEndpoint的launchTasks方法，將TaskDescription序列化並封裝在LaunchTask消息中，發送給offer指定的executor。LaunchTask消息被ExecutorBackend收到後，會將Task信息反序列化，傳給Executor.launchTask()，最後使用Executor的線程池中的線程來執行這個Task。

梳理

Stage,TaskSet,TaskSetManager是一一對應的，數量相等，都是隻存在driver上的。
Parition,Task,TaskDescription是一一對應，數量相同，Task和TaskDescription是會被髮到executor上的。

TaskScheduler的調度池

與DAGScheduler不一樣的是TaskScheduler有調度池，有兩種調度實體，Pool和TaskSetManager。
與YARN的調度隊列相似，採用了層級隊列的方式，Pool是TaskSetManager的容器，起到將TaskSetManager分組的做用。

Schedulable

Schedulable是調度實體的基類，有兩個子類Pool和TaskSetManager。

要理解調度規則，必須知道下面幾個屬性：

parent：所屬調度池，頂層的調度池爲root pool；
schedulableQueue：包含的調度對象組成的隊列；
schedulingMode：調度模式，FIFO or FAIR；
weight：權重
minShare：最小分配額(CPU核數)
runningTasks：運行中task數
priority：優先級
stageId：就是stageId
name：名稱

Pool和TaskSetManager對於這些屬性的取值有所不一樣，從而致使了他們的調度行爲也不同。

properties	Pool	TaskSetManager
weight	config	1
minShare	config	0
priority	0	jobId
stageId	-1	stageId
name	config	TaskSet_{taskSet.id}
runningTasks	Pool所含TaskSetManager的runningTasks和	TaskSetManager運行中task數

Pools建立流程

TaskScheduler有個屬性schedulingMode，值取決於配置項spark.scheduler.mode，默認爲FIFO。這個屬性會致使TaskScheduler使用不一樣的SchedulableBuilder，即FIFOSchedulableBuilder和FairSchedulableBuilder。

TaskScheduler在初始化的時候，就會建立root pool，根調度池，是全部pool的祖先。
它的屬性取值爲：

name: "" (空字符串)
schedulingMode: 同TaskScheduler的schedulingMode屬性
weight: 0
minShare: 0

注意root pool的調度模式肯定了。

接下來會執行schedulableBuilder.buildPools()方法，

若是是FIFOSchedulableBuilder，則什麼都不會發生。
如果FairSchedulableBuilder
- 1 依據scheduler配置文件(後面會說)，開始建立pool(能夠是多個pool，FIFO，FAIR都有可能，取決於配置文件)，並都加入root pool中。
- 2 若是如今root pool中沒有名爲"default"的pool(即配置文件中沒有定義一個叫default的pool)，建立default pool，並加入root pool中。

這時default pool它的屬性取值是固定的：

name: "default"
schedulingMode: FIFO
weight: 1
minShare: 0

Task加入pool流程

當TaskScheduler提交task的時候，會先建立TaskSetManager，而後經過schedulableBuilder添加到pool中。

若是是FIFOSchedulableBuilder，則會直接把TaskSetManager加入root pool隊列中。
如果FairSchedulableBuilder
- 1 從spark.scheduler.pool配置獲取pool name，沒有定義則用'default'；
- 2 從root pool遍歷找到對應名稱的pool，把TaskSetManager加入pool的隊列。若是沒有找到，則建立一個該名稱的pool，採用與default pool相同的屬性配置，並加入root pool。

調度池結構

通過上面兩部分，最終獲得的調度池結構以下：

spark.scheduler.mode=FIFO

spark.scheduler.mode=FAIR

Fair Scheduler pools配置

Fair Scheduler Pool的劃分依賴於配置文件，默認的配置文件爲'fairscheduler.xml'，也能夠經過配置項"spark.scheduler.allocation.file"指定配置文件。

煮個栗子，文件內容以下：

<?xml version="1.0"?>
<allocations>
  <pool name="prod">
    <schedulingMode>FAIR</schedulingMode>
    <weight>1</weight>
    <minShare>2</minShare>
  </pool>
  <pool name="test">
    <schedulingMode>FIFO</schedulingMode>
    <weight>2</weight>
    <minShare>3</minShare>
  </pool>
</allocations>

這裏配置了兩個pool，prod和test，而且配置了相關屬性，這兩個pool都會添加到root pool中。

調度算法

以SchedulingAlgorithm爲基類，內置實現的調度算法有兩種FIFOSchedulingAlgorithm和FairSchedulingAlgorithm，其邏輯以下：

FIFO: 先進先出，優先級比較算法以下，
- 1.比較priority，小的優先；
- 2.priority相同則比較StageId，小的優先。
FAIR：公平調度，優先級比較算法以下，
- 1.runningTasks小於minShare的優先級比不小於的優先級要高。
- 2.若二者運行的runningTasks都比minShare小，則比較minShare使用率(runningTasks/max(minShare,1))，使用率越低優先級越高。
- 3.若二者的minShare使用率相同，則比較權重使用率(runningTasks/weight)，使用率越低優先級越高。
- 4.若權重也相同，則比較name，小的優先。

Pool爲FIFO模式下的幾種情形

TaskSetManager之間的比較，其實就是先比較jobId再比較stageId，誰小誰優先，意味着就是誰先提交誰優先。

Pool之間的比較，不存在！FIFO的pool隊列中是不會有pool的。

Pool爲FAIR模式下的幾種情形

TaskSetManager之間的比較，由於minShare=0，weight=1，FAIR算法變成了：

1 runningTasks小的優先
2 runningTasks相同則比較name

Pool之間的比較，就是標準的FAIR算法。

當root pool爲FAIR模式，先取最優先的pool，再從pool中，按pool的調度模式取優先的TaskSetManager。

開始使用FAIR mode

啓用FAIR模式：

1 準備好fairscheduler.xml文件
2 啓動參數添加 --conf spark.scheduler.mode=FAIR
3 運行啓動命令，如spark-shell --master yarn --deploy-mode client --conf spark.scheode=FAIR

啓動後若是直接運行Job會自動提交到default pool，那麼如何提交Job到指定pool？
SparkContext.setLocalProperty("spark.scheduler.pool","poolName")

若是每次只運行一個Job，開啓FAIR模式的意義不大，那麼如何同時運行多個Job？
要異步提交Job，須要用到RDD的async action，目前有以下幾個：

countAsync
collectAsync
takeAsync
foreachAsync
foreachPartitionAsync

舉個例子：

sc.setLocalProperty("spark.scheduler.pool","test")
b.foreachAsync(_=>Thread.sleep(100))
sc.setLocalProperty("spark.scheduler.pool","production")
b.foreachAsync(_=>Thread.sleep(100))

這樣就會有兩個任務在不一樣的pool同時運行：