Spark RPC之Worker實作

概要

上一篇文章Spark RPC之Master實作介紹了standalone模式下Master端的實作，接着我們看下Worker端的實作，以及Worker如何向Master注冊資訊及發送心跳。

Worker

檢視Worker，Worker也是RpcEndpoint的子類，是以接下來檢視RpcEndpoint生命周期的四個方法: onStart -> receive(receiveAndReply)* -> onStop。

onStart

1. 如果配置spark.shuffle.service.enabled=true，啟動獨立的shuffle service。
2. 啟動Worker的webUI，預設端口8081。

   

3. 本篇的第一個重要部分，向Master注冊，檢視registerWithMaster方法

   

   如圖中注釋，調用tryRegisterAllMasters注冊，檢視tryRegisterAllMasters方法

   

   218行，傳回和Master通信的EndpointRef，masterEndpoint，同時調用帶參registerWithMaster方法，檢視registerWithMaster方法

   

   如上圖注釋所示，至此我們理清了onStart方法中Worker是如何注冊，同時結合上一篇Spark RPC之Master實作中Master的receiveAndReply方法對收到資訊的處理，Worker注冊的完整流程就跑完了。

   上圖第二處注釋，調用了handleRegisterResponse處理傳回資訊，這部分和心跳(Heartbeat)有關，檢視handleRegisterResponse方法

   

   傳回的消息類型有三種RegisteredWorker、RegisterWorkerFailed和MasterInStandby，第一種為注冊成功，成功後大緻操作如下
   1. 将registered屬性設定為true，修改用于和Master通信的masterRef及masterWebUiUrl。
   2. 發送心跳，本篇的第二個重點，使用了send方法發送SendHeartbeat，對應的，檢視Worker的receive方法

      

      

      最終調用masterRef的send方法将Heartbeat資訊發送給Master，對應的檢視上一篇Spark RPC之Master實作中Master的receive方法如何處理接收的Heartbeat資訊，最後，Master每60s檢視Worker連接配接情況，Worker端每15s發送一次心跳，如下

      

   3. 如果設定spark.worker.cleanup.enabled=true，清除Worker的工作目錄。

receive

SendHeartbeat是上面剛講過的，其他限于篇幅不再講述。

receiveAndReply

隻處理Worker狀态查詢。

onStop

相比于Master，多了executors、drivers和shuffleService的關閉。

Main

啟動程式和Master幾乎一緻，但至少傳入一個參數，MasterURL，格式為spark://host:port。

總結

上一篇Spark RPC之Master實作講述了Master是如何接受Worker請求注冊的資訊和心跳機制，本篇文章講解了Worker端對應的行為，如下

1. Worker的onStart方法中如何發送注冊資訊給Master

2. 在注冊成功後，Worker在處理Master傳回資訊時，啟動定時任務，每15s發送心跳

完整流程如下

至此，standalone模式下，spark如何使用rpc完成Worker注冊和心跳機制就介紹完了。