Spark Executor工作原理

ExecutorBackend注冊

Executor實(shí)例化

Executor具體是如何工作的？

?

CoarseGrainedExecutorBackend，粗粒度的ExecutorBackend進(jìn)程。

Worker為什么要啟動(dòng)另外一個(gè)進(jìn)程？

Worker本身是管理當(dāng)前機(jī)器上的資源，變動(dòng)資源的時(shí)候向Master匯報(bào)。有很多應(yīng)用程序，就需要很多Executor。這樣程序之間不會(huì)一個(gè)奔潰導(dǎo)致所有的都奔潰。

1.在CoarseGrainedExecutorBackend啟動(dòng)時(shí)，向Driver注冊Executor其實(shí)質(zhì)是注冊ExecutorBackend實(shí)例，和Executor實(shí)例之間沒有直接的關(guān)系！！！

2.CoarseGrainedExecutorBackend是Executor運(yùn)行所在的進(jìn)程名稱，Executor才是真正在處理Task的對象，Executor內(nèi)部是通過線程池的方式來完成Task的計(jì)算的。

3. CoarseGrainedExecutorBackend和Executor是一一對應(yīng)的。

4. CoarseGrainedExecutorBackend是一個(gè)消息通信體（其實(shí)現(xiàn)了ThreadSafeRpcEndpoint）。可以發(fā)送信息給Driver，并可以接收Driver中發(fā)過來的指令，例如啟動(dòng)Task等。

5.在Driver進(jìn)程中，有兩個(gè)至關(guān)重要的Endpoint，

a）第一個(gè)就是ClientEndpoint，主要負(fù)責(zé)向Master注冊當(dāng)前的程序；是AppClient的內(nèi)部成員。

b）另外一個(gè)就是DriverEndpoint，這是整個(gè)程序運(yùn)行時(shí)候的驅(qū)動(dòng)器！！是CoarseGrainedExecutorBackend的內(nèi)部成員。

6.在Driver中通過ExecutorData封裝并注冊ExecutorBackend的信息到Driver的內(nèi)存數(shù)據(jù)結(jié)構(gòu)ExecutorMapData中。ExecutorMapData是CoarseGrainedSchedulerBackend的成員。最終是注冊給CoarseGrainedSchedulerBackend。

private[cluster]?class?ExecutorData(
???val?executorEndpoint: RpcEndpointRef,
???val?executorAddress: RpcAddress,
???override val?executorHost:?String,
???var?freeCores:?Int,
???override val?totalCores:?Int,
???override val?logUrlMap:?Map[String,?String]
)?extends?ExecutorInfo(executorHost,?totalCores,?logUrlMap)

?

7.實(shí)際在執(zhí)行的時(shí)候，DriverEndpoint會(huì)把信息寫入CoarseGrainedSchedulerBackend的內(nèi)存數(shù)據(jù)結(jié)構(gòu)ExecutorMapData中，所以說最終是注冊給CoarseGrainedSchedulerBackend，也就是說CoarseGrainedSchedulerBackend掌握了為當(dāng)前程序分配的所有的ExucutorBackend進(jìn)程，而在每一個(gè)ExecutorBackend進(jìn)程實(shí)例中會(huì)通過Executor對象來負(fù)責(zé)具體Task的運(yùn)行。在欲行的時(shí)候使用syschronized關(guān)鍵字來保證ExecutorMapData安全的并發(fā)寫操作。

8.CoarseGrainedExecutorBackend收到DriverEndpoint發(fā)送過來的RegisteredExecutor消息后會(huì)啟動(dòng)Executor實(shí)例對象，而Executor實(shí)例對象是事實(shí)上負(fù)責(zé)真正Task計(jì)算的。

?

Executor是如何工作的？

1.當(dāng)Driver發(fā)送過來Task的時(shí)候，其實(shí)是發(fā)送給了CoarseGrainedExecutorBackend這個(gè)RpcEndpoint，而不是直接發(fā)送給了Executor（Executor由于不是消息循環(huán)體，所以永遠(yuǎn)無法直接接收遠(yuǎn)程發(fā)送過來的信息）。

case?LaunchTask(data) =>
??if?(executor?==?null) {
????logError("Received LaunchTask command but executor was null")
????System.exit(1)
??}?else?{
????val?taskDesc =?ser.deserialize[TaskDescription](data.value)
????logInfo("Got assigned task "?+ taskDesc.taskId)
????executor.launchTask(this,?taskId = taskDesc.taskId,?attemptNumber = taskDesc.attemptNumber,
??????taskDesc.name,?taskDesc.serializedTask)
??}

?

2.ExecutorBackend在收到Driver中發(fā)送過來的消息后，會(huì)通過調(diào)用launchTask來交給Executor去執(zhí)行。

case?LaunchTask(data) =>
??if?(executor?==?null) {
????logError("Received LaunchTask command but executor was null")
????System.exit(1)
??}?else?{
????val?taskDesc =?ser.deserialize[TaskDescription](data.value)
????logInfo("Got assigned task "?+ taskDesc.taskId)
????executor.launchTask(this,?taskId = taskDesc.taskId,?attemptNumber = taskDesc.attemptNumber,
??????taskDesc.name,?taskDesc.serializedTask)
??}

??

轉(zhuǎn)載于:https://www.cnblogs.com/haoyy/p/6177974.html

《新程序員》：云原生和全面數(shù)字化實(shí)踐50位技術(shù)專家共同創(chuàng)作，文字、視頻、音頻交互閱讀

總結(jié)

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

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