概要
上一篇文章Spark RPC之Master实现介绍了standalone模式下Master端的实现,接着我们看下Worker端的实现,以及Worker如何向Master注册信息及发送心跳。
Worker
查看Worker,Worker也是RpcEndpoint的子类,所以接下来查看RpcEndpoint生命周期的四个方法: onStart -> receive(receiveAndReply)* -> onStop。
onStart
override def onStart() {
registerWithMaster()
}
本篇的第一个重要部分,向Master注册,查看registerWithMaster方法
如图中注释,调用tryRegisterAllMasters注册,查看tryRegisterAllMasters方法
private def tryRegisterAllMasters(): Array[JFuture[_]] = {
masterRpcAddresses.map { masterAddress =>
registerMasterThreadPool.submit(new Runnable {
override def run(): Unit = {
val masterEndpoint = rpcEnv.setupEndpointRef(masterAddress, Master.ENDPOINT_NAME)
sendRegisterMessageToMaster(masterEndpoint)
}
})
}
}
val masterEndpoint = rpcEnv.setupEndpointRef(masterAddress, Master.ENDPOINT_NAME)
masterEndpoint 持有 返回的 RpcEndpointRef 类型对象的 引用,查看 sendRegisterMessageToMaster 方法:
private def sendRegisterMessageToMaster(masterEndpoint: RpcEndpointRef): Unit = {
masterEndpoint.send(RegisterWorker(...))
}
调用了 RpcEndpointRef 的 send 方法,把 RegisterWorker 类型消息发送到 Master,Master 收到 Worker 的 RegisterWorker 消息:
override def receive: PartialFunction[Any, Unit] = {
case RegisterWorker(
val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,
workerRef, workerWebUiUrl)
if (registerWorker(worker)) {
persistenceEngine.addWorker(worker)
//发送 RegisteredWorker 消息给 Worker
workerRef.send(RegisteredWorker(self, masterWebUiUrl, masterAddress))
schedule()
}
}
Worker 在 receive 方法中接受消息:
override def receive: PartialFunction[Any, Unit] = synchronized {
case msg: RegisterWorkerResponse =>
handleRegisterResponse(msg)
}
Worker 收到 RegisterWorkerResponse 类型消息,则调用 handleRegisterResponse 方法:
private def handleRegisterResponse(msg: RegisterWorkerResponse): Unit = synchronized {
msg match {
case RegisteredWorker(masterRef, masterWebUiUrl, masterAddress) =>
registered = true
forwordMessageScheduler.scheduleAtFixedRate(new Runnable {
override def run(): Unit = Utils.tryLogNonFatalError {
self.send(SendHeartbeat)
}
}, 0, HEARTBEAT_MILLIS, TimeUnit.MILLISECONDS)
case RegisterWorkerFailed(message) =>
case MasterInStandby =>
}
}
返回的消息类型有三种RegisteredWorker、RegisterWorkerFailed和MasterInStandby,这三种消息都继承自:RegisterWorkerResponse,所以可以在 receive 中匹配成功。
在 handleRegisterResponse 方法中,匹配到 Master 发送的 RegisteredWorker 消息,并最终调用 self 的 send(SendHeartbeat) 发送给你自己,Worker自己接收并处理:
case SendHeartbeat =>
if (connected) { sendToMaster(Heartbeat(workerId, self)) }
sendToMaster 方法:
private def sendToMaster(message: Any): Unit = {
master match {
case Some(masterRef) => masterRef.send(message)
}
}
还是调用 RpcEndpointRef send方法,发送 Heartbeat 消息 到 Master,Master receive 中接收:
case Heartbeat(workerId, worker) =>
idToWorker.get(workerId) match {
case Some(workerInfo) =>
workerInfo.lastHeartbeat = System.currentTimeMillis()
}
最后,Master每60s查看Worker连接情况,Worker端每15s发送一次心跳,如下
receive
SendHeartbeat是上面刚讲过的,其他限于篇幅不再讲述。
receiveAndReply
只处理Worker状态查询。
onStop
相比于Master,多了executors、drivers和shuffleService的关闭。
Main
启动程序和Master几乎一致
总结
上一篇Spark RPC之Master实现讲述了Master是如何接受Worker请求注册的信息和心跳机制,本篇文章讲解了Worker端对应的行为,如下
1. Worker的onStart方法中如何发送注册信息给Master
2. 在注册成功后,Worker在处理Master返回信息时,启动定时任务,每15s发送心跳
完整流程如下
至此,standalone模式下,spark如何使用rpc完成Worker注册和心跳机制就介绍完了。
网友评论