【Spark2.0源碼學習】-5.Worker啓動

     Worker做爲Endpoint的具體實例，下面咱們介紹一下Worker啓動以及OnStart指令後的額外工做

 

1、腳本概覽

     下面是一個舉例：

/opt/jdk1.7.0_79/bin/java
-cp /opt/spark-2.1.0/conf/:/opt/spark-2.1.0/jars/*:/opt/hadoop-2.6.4/etc/hadoop/
-Xmx1g
-XX:MaxPermSize=256m
org.apache.spark.deploy.worker.Worker
--webui-port 8081
spark://zqh:7077

 

2、啓動流程

     Worker的啓動流程以下：

     

• SparkConf：加載key以spark.開頭的系統屬性（Utils.getSystemProperties）
• WorkerArguments：
• • 解析Master啓動的參數（--ip -i --host -h --port -p --cores -c --memory -m --work-dir --webui-port  --properties-file）
  • 將--properties-file（沒有配置默認爲conf/spark-defaults.conf）中spark.開頭的配置存入SparkConf
  • 在沒有配置狀況下，cores默認爲服務器CPU核數
  • 在沒有配置狀況下，memory默認爲服務器內存減1G，若是低於1G取1G
  • webUiPort默認爲8081
• NettyRpcEnv中的內部處理遵循RpcEndpoint統一處理，這裏再也不贅述
• 最終守護進程會一直存在等待結束信awaitTermination

 

3、OnStart 監聽事件

     Worker的啓動完成後異步執行工做以下：

     

• 【dispatcher-event-loop】線程掃描到OnStart指令後會啓動相關WorkerWebUI(默認端口8081）
• Worker向Master發起一次RegisterWorker指令
• 另起【master-forward-message-thread】線程按期執行ReregisterWithMaster任務，若是註冊成功（RegisteredWorker）則跳過，不然再次向Master發起RegisterWorker指令，直到超過最大次數報錯（默認16次）
• Master若是能夠註冊，則維護對應的WorkerInfo對象並持久化，完成後向Worker發起一條RegisteredWorker指令，若是Master爲standby狀態，則向Worker發起一條MasterInStandby指令
• Worker接受RegisteredWorker後，提交【master-forward-message-thread】線程按期執行SendHeartbeat任務，，完成後向Worker發起一條WorkerLatestState指令
• Worker發心跳檢測，會觸發更新Master對應WorkerInfo對象，若是Master檢測到異常，則發起ReconnectWorker指令至Worker，Worker則再次執行ReregisterWithMaster工做

 

4、RpcMessage處理 (receiveAndReply)

 

消息實例	發起方	接收方	說明
RequestWorkerState	WorkerWebUI	Worker	返回 WorkerStateResponse

 

5、OneWayMessage 處理 (receive)

消息實例	發起方	接收方	說明
SendHeartbeat	Worker	Worker
WorkDirCleanup	Worker	Worker
ReregisterWithMaster	Worker	Worker
MasterChanged	Master	Worker
ReconnectWorker	Master	Worker
LaunchExecutor	Master	Worker
ApplicationFinished	Master	Worker
KillExecutor	Master	Worker
LaunchDriver	Master	Worker
KillDriver	Master	Worker
DriverStateChanged	DriverRunner	Worker
ExecutorStateChanged	ExecutorRunner /Worker	Worker/Master