目錄

?

1、Job Stage劃分

2、Task最佳位置

3、總結(jié)

3.1 Stage劃分總結(jié)：

3.2 Task最佳位置總結(jié)：

---

1、Job Stage劃分

Spark Application中因為不同的Action觸發(fā)眾多的Job，也就是說一個Application中可以有很多的Job，每個Job是由是由一個或者多個Stage構(gòu)成的，后面的Stage依賴于前面的Stage，也就是說只有前面依賴的Stage計算完畢后，后面的Stage才會運行。而Stage劃分的依據(jù)就是寬依賴。下面以RDD的collect方法為例：

（1）他是一個action會觸發(fā)一個具體的作業(yè)runJob

（2）runJob有很多重載方法，不斷地往里調(diào)用，最后交給dagScheduler的runJob，在dagScheduler的runJob交給了submitJob，后面還有一個等待作業(yè)結(jié)果看成功還是失敗，會有相應(yīng)的動作。

（3）在submitJob中首先看一下分區(qū)長度，是因為要進行計算，這個肯定是RDD導(dǎo)致的action他要校驗一下是不是在運行的時候相應(yīng)的Partition存在。

eventProcessLoop調(diào)用post的時候有個Jobsubmitted的參數(shù)，他是一個case class，因為一個application中可能有很多的Job，不同的job的Jobsubmitted實例不一樣所以不能用case object。他里面封裝了job的id，最后一個RDD，具體對RDD操作的函數(shù)，有哪些Partition要被計算，監(jiān)聽作業(yè)狀態(tài)等。

他的核心就是將Jobsubmitted交給eventProcessLoop。他是通過post方法post給eventProcessLoop，這個post其實就是發(fā)往EventLoop里面的eventQueue

（4）發(fā)現(xiàn)在EventLoop里面開辟了一個線程，他是setDaemon方式作為后臺線程，因為要在后臺做不斷的循環(huán)（如果是前臺線程的話對垃圾回收是有影響的），在run方法里面會不斷的循環(huán)我們的消息隊列，從eventQueue（是一個LinkedBlockingDeque，我們可以往他里面信息）中獲得消息，調(diào)用了onReceive，發(fā)現(xiàn)在里面沒有具體的實現(xiàn)所以在DAGSchedulerEventProcessLoop中對onReceive進行了實現(xiàn)，這里就收到了DAGSchedulerEvent，這里面再調(diào)用doOnReceive。doOnReceive收到信息就開始處理

（5）接下來就是HandleJobSubmited。這個時候Stage就開始了。我們知道最后一個Stage一定是ResultStage，前面所有的Stage都是ShuffleMapStage。

（6）發(fā)現(xiàn)有個getOrCreateParentStages的方法，開始創(chuàng)建ResultStage的父stage，里面有多個嵌套獲取shuffle依賴和循環(huán)創(chuàng)建shuffleMapStage，若沒有shuffle，操作則返回空list

進入到創(chuàng)建父Stage的方法getOrCreateParentStages，這里僅僅是抽取當前RDD的shuffle依賴，shuffleMapStage，如果不是shuffleDependency就繼續(xù)抽取父RDD，迭代遍歷一直到抽取出為止或者沒有

進入getOrCreateShuffleMapStage方法中，進行匹配能不能取到ParentStage的值，當沒有parentStage的時候會返回空，能取到就返回stage，ShuffleMapStage是根據(jù)遍歷出的ShuffleDependencies一次次創(chuàng)建出來的

進入createShuffleMapStage方法 此方法是遞歸循環(huán)創(chuàng)建shuffleMapStage的過程

這個時候ShuffleMapStage已經(jīng)創(chuàng)建完成了，并不是一次就創(chuàng)建完成，而是遇見shuffle的時候會由下往上遞歸創(chuàng)建ShuffleMapStage

（7）構(gòu)建完所有的ShuffleMapStage后，將其作為參數(shù)創(chuàng)建ResultStage

（8）最后將Stage和id關(guān)聯(lián)，更新job所有的Stage，并將Stage返回出去。

（9）回到handleJobsubmited方法中，finalStage構(gòu)建完之后，新建一個ActiveJob保存了當前job的一些信息，打印一堆日志之類。getMissingParentStages(finalStage)根據(jù)finalStage，剛才找父Stage的時候如果有的話直接返回，如果沒有的話就會創(chuàng)建，所以如果曾經(jīng)有就不需要再去做。listenerBus.post監(jiān)聽事件，最后submitStage(finalStage)。

首先獲得id，如果jobId是defined的話再次getMissingParentStages(stage)獲得missing的stage之后判斷一下是否為空，如果為空的話就submitMissingTasks(stage, jobId.get)個就是沒有前置性的Tasks，也就是沒有父Stage。在這個底層其實是DAGScheduler把這個處理的過程交給具體的TaskScheduler去處理

2、Task最佳位置

（1）在handleJobsubmited方法中最后是最后調(diào)用submitStage，在他里面會調(diào)用submitMissingTasks

（2）這里面有很多代碼，我們要關(guān)心Stage本身的算法以及Task任務(wù)本地性把當前的Stage加進去，然后對Stage進行判斷，一種是ShuffleMapStage，一種是ResultStage。繼續(xù)往下走會看到taskIdToLocations這是關(guān)鍵的代碼，taskIdToLocations是一個Map

partitionsToCompute這里面獲得是具體的要計算的PartitionID，我們我們這邊看到的map里面的id是Partition的id。這里面匿名函數(shù)，產(chǎn)生的是tuple根據(jù)Partition的id。后面toMap就是Partition的id和TaskLocation的位置。

（3）進入到getPreferredLocs(stage.rdd, id)，進來的是RDD，PartitionID返回的是一個集合。

再進入getPreferredLocsInternal

visited: HashSet[(RDD[_], Int)]這個HashSet開始是空，所以直接傳進來一個new HashSet，然后判斷visited如果已經(jīng)有的話，那么添加就不成功，那么就是已經(jīng)計算了數(shù)據(jù)本地性了，就返回Nil。

下面的cached就是已經(jīng)在DAGScheduler的內(nèi)存數(shù)據(jù)結(jié)構(gòu)中了。進入getCacheLocs，這邊返回的是序列，cacheLocs是一個HashMap，這包含了每個RDD的Partition的id以及id對應(yīng)的taskLocation，這個包含了Stage本身也包含了Stage內(nèi)部任務(wù)的本地性

（4）回到getPreferredLocsInternal中，上面是看一下DAGScheduler中有沒有緩存根據(jù)Partition而保存的數(shù)據(jù)本地性的內(nèi)容，如果不為空的話就把內(nèi)容返回。然后調(diào)用下面的getpreferdLocations（如果自定義一個RDD的話是一定要寫這個方法的）

（5）最后判斷一下如果是窄依賴的話就自己調(diào)用自己

3、總結(jié)

3.1 Stage劃分總結(jié)：

（1）Action觸發(fā)Job，開始逆向分析job執(zhí)行過程Action中利用SparkContext runJob路由到dagScheduler.runJob(rdd,func,分區(qū)數(shù)，其他)，提交Job作業(yè)；

（2）DAGScheduler的runJob中調(diào)用submitJob并返回監(jiān)聽waiter，生命周期內(nèi)監(jiān)聽Job狀態(tài)；

（3）在submitJob內(nèi)部，將該獲取到的Job(已有JobId），插入到名為eventProcessLoop的LinkedBlockingDeque結(jié)構(gòu)的事件處理隊列中；

（4）eventProcessLoop放入新事件后，調(diào)起底層的DAGSchedulerEventProcessLoop的onReceive方法；

（5）執(zhí)行doOnReceive，根據(jù)DAGSchedulerEvent的具體類型如JobSubmitted事件或者MapStageSubmitted事件，調(diào)取具體的Submitted handle函數(shù)提交具體的Job；

（6）以JobSubmitted為例，在handleJobSubmitted內(nèi)部，返回從ResultStage 建立stage 建立finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite)，finalStage激活Job val job = new ActiveJob(jobId, finalStage, callSite, listener, properties)，同時開始逆向構(gòu)建缺失的stage；

（7）DAG構(gòu)建完畢，提交stage，submitStage(finalStage)，submitStage中stage提交為tasks，submitMissingTasks（），submitMissingTasks，根據(jù)ShuffleMapStage還是ResultStage創(chuàng)建 ShuffleMapTask 或 ResultTask。

（7）taskScheduler.submitTasks（）開始調(diào)起具體的task

3.2 Task最佳位置總結(jié)：

（1）在劃分Stage的時候submitMissingTasks方法中會有一個taskIdToLocations的屬性，他的結(jié)構(gòu)為 Map[Int, Seq[TaskLocation]]，他保存的就是PartitionID及其對應(yīng)的最佳位置

（2）在對taskIdToLocations賦值的時候會調(diào)用getPreferredLocs方法，再路由到getPreferredLocsInternal返回最佳位置Seq[TaskLocation]

（3）在getPreferredLocsInternal方法中

①判斷rdd的partition是否被訪問過，如果被訪問過，則什么都不做

②然后判斷DAGScheduler的內(nèi)存中是否cache了在當前Paritition的信息，如果有的話直接返回

③如果沒有cache，則調(diào)用rdd.getPreferredLocations方法，獲取RDD partition的最佳位置

④遍歷RDD的依賴，如果有窄依賴，遍歷父依賴的partition，對遍歷到的每個partition，遞歸調(diào)用getPreferredLocsInternal方法

即從第一個窄依賴的第一個partition開始，然后將每個partition的最佳位置，添加到序列中，最后返回所有partition的最佳位置序列

注意：DAGScheduler計算數(shù)據(jù)本地性的時候借助了RDD自身的getPreferredLocations中的數(shù)據(jù)，因為getPreferredLocations中表明了每個Partition的數(shù)據(jù)本地性，雖然當前Partition可能被persist或者checkpoint，但是persist或者checkpoint默認情況下肯定是和getPreferredLocations中的Partition的數(shù)據(jù)本地性是一致的，所以這就極大的簡化Task數(shù)據(jù)本地性算法的實現(xiàn)和效率的優(yōu)化。

總結(jié)

以上是生活随笔為你收集整理的Stage划分和Task最佳位置的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。