目錄

?

1、Spark Executor如何工作

2、Spark Executor工作源碼

---

1、Spark Executor如何工作

當Driver發(fā)送過來Task的時候，其實是發(fā)送給CoarseGrainedExecutorBackend這個RPCEndpoint，而不是直接發(fā)送給Executor（Executor由于不是消息循環(huán)體永遠也無法接收遠程發(fā)過來的信息）。

Driver向ExecutorBackend發(fā)送LaunchTask，這里實際上是把內(nèi)容交給線程池中的線程去執(zhí)行。首先判斷Executor是否為空，反序列化TaskDescription，然后調(diào)用Executor.launchTask ，launchTask 里面是將Task封裝在TaskRunner（是一個runnable對象）里面然后交給線程池中的線程處理。在TaskRunner的run方法里面會導致runTask的執(zhí)行。

2、Spark Executor工作源碼

（1）Worker接收到Master發(fā)送過來啟動Executor的指令，通過ExecutorRunner啟動另外一個進程來運行Executor。在這個基礎上會啟動一個CoarseGrainedExecutorBackend（粗粒度），ExecutorBackend啟動的時候這個時候要向Driver注冊，通過RegisterExecutor。Driver（CoarseGrainedSchedulerBackend的內(nèi)部成員DriverEndpoint接收注冊）在接收到RegisterExecutor信息后會返回一個信息RegisteredExecutor給CoarseGrainedExecutorBackend。

CoarseGrainedExecutorBackend：是Executor運行所在的進程的名稱，他本身不會完成具體任務的計算，這個進程里面有個Executor對象和CoarseGrainedExecutorBackend是一一對應的。

Executor：完成具體的計算，真正處理Task的對象，內(nèi)部通過線程池的方式完成Task的計算。

（2）Driver的CoarseGrainedSchedulerBackend內(nèi)部的成員DriverEndpoint收到RegisterExecutor后，首先判斷ExecutorDataMap中是否有這個ExecutorId。ExecutorDataMap是個內(nèi)存數(shù)據(jù)結構（他是CoarseGrainedSchedulerBackend的成員，所以最終是注冊給CoarseGrainedSchedulerBackend），他是一個HashMap。

如果ExecutorDataMap中不存在，就進行注冊，注冊的時候先看看address存在不存在，如果存在直接獲取，如果不存在就獲取senderAddress。這從實現(xiàn)的角度看adress就是senderAddress。然后相關數(shù)據(jù)結構添加數(shù)據(jù)

CoarseGrainedSchedulerBackend.this.synchronized這里加上synchronized是因為集群中有很多ExecutorBackend向Driver注冊，擔心注冊的時候?qū)憶_突，所以加上一個同步代碼塊

最后把消息發(fā)還給CoarseGrainedExecutorBackend

（3）CoarseGrainedExecutorBackend收到RegisterExecutor后就new了一個Executor這個Executor對象是事實上負責Task計算的

（4）在Executor默認構造器中有一個非常關鍵的內(nèi)容，有個成員threadPool（線程池）

創(chuàng)建線程池，線程池里面要有線程，線程怎么產(chǎn)生的呢？不會平白無故的產(chǎn)生，所以就搞了一個線程工廠，就是按照某種你需要的格式去產(chǎn)生線程，背后還是new出一個線程

setDamon(True)設置每個線程是后臺運行的方式

（5）線程池也準備好后就是等待Driver端發(fā)任務過來，是發(fā)給CoarseGrainedExecutorBackend不是Executor，因為Executor不可能接收到消息的他本身就不是一個消息循環(huán)體。CoarseGrainedExecutorBackend收到LaunchTask后，這里實際上是把內(nèi)容交給線程池中的線程去執(zhí)行。判斷Executor是否為空，反序列化TaskDescription，然后調(diào)用Executor.launchTask

（6）launchTask 里面是將Task封裝在TaskRunner（是一個runnable對象）里面然后交給線程池中的線程處理。并把這個任務加入ConcurrentHashMap類型的名稱為runningTasks的數(shù)據(jù)結構中管理。

（7）在TaskRunner的run方法里面會導致runTask的執(zhí)行

補充：為什么要在worker接收到master發(fā)送過來的指令后為什么要啟動另外一個進程，也就是說為什么開辟另外一個進程，在另外一個進程中注冊給Driver，然后啟動Executor。必須啟動另外一個進程的原因：①Worker本身是管理當前機器上的資源的，當前機器上的資源變動的時候要匯報給Master，Worker不是用來做計算的不能在Worker里面計算；②Spark集群中可能有很多應用程序就可能有很多的Executor，如果不是為每個Executor啟動一個進程而是所有Executor在Worker里面，那么一個程序奔潰了會導致其他程序奔潰。

?

?

?

?

總結

以上是生活随笔為你收集整理的Spark Executor解析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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