Spark内核解析2

港控/mmm° 2022-01-17 17:57 199阅读 0赞

Spark的交互流程- - -节点启动：

![20190609154253.png][]

Master启动时首先创一个RpcEnv对象，负责管理所有通信逻辑  
Master通过RpcEnv对象创建一个Endpoint，Master就是一个Endpoint，Worker可以与其进行通信  
Worker启动时也是创一个RpcEnv对象  
Worker通过RpcEnv对象创建一个Endpoint  
Worker通过RpcEnv对，建立到Master的连接，获取到一个RpcEndpointRef对象，通过该对象可以与Master通信  
Worker向Master注册，注册内容包括主机名、端口、CPU Core数量、内存数量  
Master接收到Worker的注册，将注册信息维护在内存中的Table中，其中还包含了一个到Worker的RpcEndpointRef对象引用  
Master回复Worker已经接收到注册，告知Worker已经注册成功  
此时如果有用户提交Spark程序，Master需要协调启动Driver；而Worker端收到成功注册响应后，开始周期性向Master发送心跳

## Mster ##

Master的main方法启动创建了一个startRpcEnvAndEndpoint

我们点击`startRpcEnvAndEndpoint`查看中创建了一个Master

在Master中我们重点关注一下onStart()方法,绑定WebUI,和定期心跳检测

我们在查看一下receive 这些只是受到消息,然而并没有恢复

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 </pre></td> 
 <td><pre>override def receive: PartialFunction[Any, Unit] = {
 case ElectedLeader =&gt; // Leader选举,用在SparkHA中 val (storedApps, storedDrivers, storedWorkers) = persistenceEngine.readPersistedData(rpcEnv) state = if (storedApps.isEmpty &amp;&amp; storedDrivers.isEmpty &amp;&amp; storedWorkers.isEmpty) {
 RecoveryState.ALIVE } else {
 RecoveryState.RECOVERING } logInfo("I have been elected leader! New state: " + state) if (state == RecoveryState.RECOVERING) {
 beginRecovery(storedApps, storedDrivers, storedWorkers) recoveryCompletionTask = forwardMessageThread.schedule(new Runnable {
 override def run(): Unit = Utils.tryLogNonFatalError {
 self.send(CompleteRecovery) } }, WORKER_TIMEOUT_MS, TimeUnit.MILLISECONDS) } case CompleteRecovery =&gt; completeRecovery() //成功地恢复 case RevokedLeadership =&gt; // logError("Leadership has been revoked -- master shutting down.") System.exit(0) 	//注册应用 case RegisterApplication(description, driver) =&gt; // TODO Prevent repeated registrations from some driver if (state == RecoveryState.STANDBY) {
 // ignore, don't send response } else {
 logInfo("Registering app " + description.name) val app = createApplication(description, driver) registerApplication(app) logInfo("Registered app " + description.name + " with ID " + app.id) persistenceEngine.addApplication(app) driver.send(RegisteredApplication(app.id, self)) // 启动分配Executor schedule() } // 检测 Executor case ExecutorStateChanged(appId, execId, state, message, exitStatus) =&gt; val execOption = idToApp.get(appId).flatMap(app =&gt; app.executors.get(execId)) execOption match {
 case Some(exec) =&gt; val appInfo = idToApp(appId) val oldState = exec.state exec.state = state if (state == ExecutorState.RUNNING) {
 assert(oldState == ExecutorState.LAUNCHING, s"executor $execId state transfer from $oldState to RUNNING is illegal") appInfo.resetRetryCount() } exec.application.driver.send(ExecutorUpdated(execId, state, message, exitStatus, false)) if (ExecutorState.isFinished(state)) {
 // Remove this executor from the worker and app logInfo(s"Removing executor ${exec.fullId} because it is $state") // If an application has already finished, preserve its // state to display its information properly on the UI if (!appInfo.isFinished) {
 appInfo.removeExecutor(exec) } exec.worker.removeExecutor(exec) val normalExit = exitStatus == Some(0) // Only retry certain number of times so we don't go into an infinite loop. // Important note: this code path is not exercised by tests, so be very careful when // changing this `if` condition. if (!normalExit &amp;&amp; appInfo.incrementRetryCount() &gt;= MAX_EXECUTOR_RETRIES &amp;&amp; MAX_EXECUTOR_RETRIES &gt;= 0) { // &lt; 0 disables this application-killing path val execs = appInfo.executors.values if (!execs.exists(_.state == ExecutorState.RUNNING)) {
 logError(s"Application ${appInfo.desc.name} with ID ${appInfo.id} failed " + s"${appInfo.retryCount} times; removing it") removeApplication(appInfo, ApplicationState.FAILED) } } } schedule() case None =&gt; logWarning(s"Got status update for unknown executor $appId/$execId") } case DriverStateChanged(driverId, state, exception) =&gt; state match {
 case DriverState.ERROR | DriverState.FINISHED | DriverState.KILLED | DriverState.FAILED =&gt; removeDriver(driverId, state, exception) case _ =&gt; throw new Exception(s"Received unexpected state update for driver $driverId: $state") } 	//心跳机制 case Heartbeat(workerId, worker) =&gt; idToWorker.get(workerId) match {
 case Some(workerInfo) =&gt; workerInfo.lastHeartbeat = System.currentTimeMillis() case None =&gt; if (workers.map(_.id).contains(workerId)) {
 logWarning(s"Got heartbeat from unregistered worker $workerId." + " Asking it to re-register.") worker.send(ReconnectWorker(masterUrl)) } else {
 logWarning(s"Got heartbeat from unregistered worker $workerId." + " This worker was never registered, so ignoring the heartbeat.") } } case MasterChangeAcknowledged(appId) =&gt; idToApp.get(appId) match {
 case Some(app) =&gt; logInfo("Application has been re-registered: " + appId) app.state = ApplicationState.WAITING case None =&gt; logWarning("Master change ack from unknown app: " + appId) } if (canCompleteRecovery) { completeRecovery() } 	//反馈Worker的运行状态 case WorkerSchedulerStateResponse(workerId, executors, driverIds) =&gt; idToWorker.get(workerId) match {
 case Some(worker) =&gt; logInfo("Worker has been re-registered: " + workerId) worker.state = WorkerState.ALIVE val validExecutors = executors.filter(exec =&gt; idToApp.get(exec.appId).isDefined) for (exec &lt;- validExecutors) {
 val app = idToApp.get(exec.appId).get val execInfo = app.addExecutor(worker, exec.cores, Some(exec.execId)) worker.addExecutor(execInfo) execInfo.copyState(exec) } for (driverId &lt;- driverIds) {
 drivers.find(_.id == driverId).foreach { driver =&gt; driver.worker = Some(worker) driver.state = DriverState.RUNNING worker.drivers(driverId) = driver } } case None =&gt; logWarning("Scheduler state from unknown worker: " + workerId) } if (canCompleteRecovery) { completeRecovery() } case WorkerLatestState(workerId, executors, driverIds) =&gt; idToWorker.get(workerId) match {
 case Some(worker) =&gt; for (exec &lt;- executors) {
 val executorMatches = worker.executors.exists {
 case (_, e) =&gt; e.application.id == exec.appId &amp;&amp; e.id == exec.execId } if (!executorMatches) {
 // master doesn't recognize this executor. So just tell worker to kill it. worker.endpoint.send(KillExecutor(masterUrl, exec.appId, exec.execId)) } } for (driverId &lt;- driverIds) {
 val driverMatches = worker.drivers.exists { case (id, _) =&gt; id == driverId } if (!driverMatches) {
 // master doesn't recognize this driver. So just tell worker to kill it. worker.endpoint.send(KillDriver(driverId)) } } case None =&gt; logWarning("Worker state from unknown worker: " + workerId) } case UnregisterApplication(applicationId) =&gt; logInfo(s"Received unregister request from application $applicationId") idToApp.get(applicationId).foreach(finishApplication) 	//给积极扫那个一个消息,检测自己是否挂掉 case CheckForWorkerTimeOut =&gt; timeOutDeadWorkers() } </pre></td> 
 </tr> 
 </tbody> 
</table>

注意:上述都是没有回复的,我们需要查看一下`receiveAndReply`

## Worker ##

我们查看一下Worker的main方法

我们在查看一下调用的`startRpcEnvAndEndpoint`

我们在查看一下`onStar`t方法

我们查看一下`registerWithMaster`

点击`ReregisterWithMaster`

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 </pre></td> 
 <td><pre>case ReregisterWithMaster =&gt; reregisterWithMaster() // 调用方法 </pre></td> 
 </tr> 
 </tbody> 
</table>

## Master ##

在Master中我们查看一下`receiveAndReply`

点击查看`registerWorker`

在Master上如果我们成功了则调用`context.reply(RegisteredWorker(self, masterWebUiUrl))` 告知Worker注册成功

点击进入`sendToMaster`

Master中收到发送的消息

我们回过头看一下Master中的`Onstart`调用的`self.send(CheckForWorkerTimeOut)`

<table> 
 <tbody> 
 <tr> 
 <td><pre>1 2 </pre></td> 
 <td><pre>case CheckForWorkerTimeOut =&gt; timeOutDeadWorkers() //调用 </pre></td> 
 </tr> 
 </tbody> 
</table>

[20190609154253.png]: /images/20220114/7de959578497412886fc7fe801cf983b.png