Spark2.0源码学习-5.Worker启动

Posted 2020-09-15

     Worker作为Endpoint的具体实例，下面我们介绍一下Worker启动以及OnStart指令后的额外工作

 

一、脚本概览

     下面是一个举例：

/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

 

二、启动流程

     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

 

三、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工作

 

四、RpcMessage处理(receiveAndReply)

 

消息实例	发起方	接收方	说明
RequestWorkerState	WorkerWebUI	Worker	返回 WorkerStateResponse

 

五、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