【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