ideaspark源碼

『壹』 Spark源碼分析之SparkSubmit的流程

本文主要對SparkSubmit的任務提交流程源碼進行分析。 Spark源碼版本為2.3.1。

首先閱讀一下啟動腳本，看看首先載入的是哪個類，我們看一下 spark-submit 啟動腳本中的具體內容。

可以看到這里載入的類是org.apache.spark.deploy.SparkSubmit，並且把啟動相關的參數也帶過去了。下面我們跟一下源碼看看整個流程是如何運作的...

SparkSubmit的main方法如下

這里我們由於我們是提交作業，所有會走上面的submit(appArgs, uninitLog)方法

可以看到submit方法首先會准備任務提交的環境，調用了prepareSubmitEnvironment，該方法會返回四元組，該方法中會調用doPrepareSubmitEnvironment，這里我們重點注意 childMainClass類具體是什麼 ，因為這里涉及到後面啟動我們主類的過程。

以下是doPrepareSubmitEnvironment方法的源碼...

可以看到該方法首先是解析相關的參數，如jar包，mainClass的全限定名，系統配置，校驗一些參數，等等，之後的關鍵點就是根據我們 deploy-mode 參數來判斷是如何運行我們的mainClass，這里主要是通過childMainClass這個參數來決定下一步首先啟動哪個類。

childMainClass根據部署模型有不同的值:

之後該方法會把准備好的四元組返回，我們接著看之前的submit方法

可以看到這里最終會調用doRunMain()方法去進行下一步。

doRunMain的實現如下...

doRunMain方法中會判斷是否需要一個代理用戶，然後無論需不需要都會執行runMain方法，我們接下來看看runMain方法是如何實現的。

這里我們只假設以集群模式啟動，首先會載入類，將我們的childMainClass載入為位元組碼對象mainClass ，然後將mainClass 映射成SparkApplication對象，因為我們以集群模式啟動，那麼上一步返回四元組中的childMainClass的參數為ClientApp的全限定名，而這里會調用app實例的start方法因此，這里最終調用的是ClientApp的start方法。

ClientApp的start方法如下...

可以看到這里和之前我們的master啟動流程有些相似。
可以參考我上一篇文章 Spark源碼分析之Master的啟動流程 對這一流程加深理解。

首先是准備rpcEnv環境，之後通過master的地址獲取masterEndpoints端點相關信息，因為這里運行start方法時會將之前配置的相關參數都傳進來，之後就會通過rpcEnv注冊相關clientEndPoint端點信息，同時需要注意，這里會把masterEndpoints端點信息也作為構造ClientEndpoint端點的參數，也就是說這個ClientEndpoint會和masterEndpoints通信。

而在我上一篇文章中說過，只要是setupEndpoint方法被調用，一定會調用相關端點的的onStart方法，而這會調用clientEndPoint的onStart方法。

ClientEndPoint類中的onStart方法會匹配launch事件。源碼如下

onStart中匹配我們的launch的過程，這個過程是啟動driverWrapper的過程，可以看到上面源碼中封裝了mainClass ，該參數對應DriverWrapper類的全限定名，之後將mainClass封裝到command中，然後封裝到driverDescription中，向Master申請啟動Driver。

這個過程會向Mster發送消息，是通過rpcEnv來實現發射消息的，而這里就涉及到outbox信箱，會調用postToOutbox方法，向outbox信箱中添加消息，然後通過TransportClient的send或sendRpc方法發送消息。發件箱以及發送過程是在同一個線程中進行。

而細心的同學會注意到這里調用的方法名為SendToMasterAndForwardReply，見名之意，發送消息到master並且期待回應。

下面是rpcEnv來實現向遠端發送消息的一個調用流程，最終會通過netty中的TransportClient來寫出。

之後，Master端會觸發receiveAndReply函數,匹配RequestSubmitDriver樣例類，完成模式匹配執行後續流程。

可以看到這里首先將Driver信息封裝成DriverInfo，然後添加待調度列表waitingDrivers中，然後調用通用的schele函數。

由於waitingDrivers不為空，則會走LaunchDriver的流程，當前的application申請資源，這時會向worker發送消息，觸發Worker的receive方法。

Worker的receive方法中，當Worker遇到LaunchDriver指令時，創建並啟動一個DriverRunner，DriverRunner啟動一個線程，非同步的處理Driver啟動工作。這里說啟動的Driver就是剛才說的org.apache.spark.deploy.worker.DriverWrapper

可以看到上面在DriverRunner中是開辟線程非同步的處理Driver啟動工作，不會阻塞主進程的執行，而prepareAndRunDriver方法中最終調用 runDriver..

runDriver中主要先做了一些初始化工作，接著就開始啟動driver了。

上述Driver啟動工作主要分為以下幾步：

下面我們直接看DriverWrapper的實現

DriverWrapper，會創建了一個RpcEndpoint與RpcEnv，RpcEndpoint為WorkerWatcher，主要目的為監控Worker節點是否正常，如果出現異常就直接退出，然後當前的ClassLoader載入userJar，同時執行userMainClass，在執行用戶的main方法後關閉workerWatcher。

以上就是SparkSubmit的流程，下一篇我會對SparkContext的源碼進行解析。

歡迎關注...