升级ExecutorBackend通信机制简述

在ExecutorBackend中有statusUpdate（taskId：Long，state：TaskState，data：ByteBuffer）方法，通过这个方法向集群发送Task执行的各种信息。如果任务执行失败，返回失败的信息；如果执行成功，返回任务执行的结果。在本节中，将重点讲解在Standalone模式下，Coar-seGrainedExecutorBackend中的通信。CoarseGrainedExecutorBackend在整个集群中的通信如图5-4所示。

在图5-4中，Executor与CoarseGrainedExecutorBackend协作，任务计算的结果通过CoarseGrainedExecutorBackend的statusUpdate方法将taskId、TaskState及结果数据发送给Driver。Driver收到StatusUpdate（executorId，tasked，state，data）消息，通过判断state的不同的状态来进行不同的处理。比如，当state的状态为TaskState.LOST时，Driver端将会移除Executor；当state状态为TaskState.FINISHED时，Driver端将会调用enqueueSuccessfulTask进行处理。

首先看一下CoarseGrainedExecutorBackend与Driver之间的通信。当在Worker结点中启动ExecutorRunner时，ExecutorRunner中会启动CoarseGrainedExecutorBackend进程，在Coar-

图5-4 CoarseGrainedExecutorBackend在整个集群中的通信

seGrainedExecutorBackend的onStart方法中，向Driver发出RegisterExecutor注册请求。源代码如下所示。

上面代码中，Some（ref）得到Driver的引用，通过ask方法返回Future［RegisterExecutor-Response］，然后在Future对象上调用onComplete方法进行额外的处理。Driver端收到注册请求，将会注册Executor的请求，并向ListenerBus中发送SparkListenerExecutorAdded事件。处理源代码如下所示。

(www.xing528.com)

如上面代码所示，如果executorDataMap中已经存在该Executor的id，返回RegisterExec-utorFailed；如果不存在该Executor的id，则在executorDataMap中加入该Executor的id，并返回RegisteredExecutor消息且向listenerBus中添加SparkListenerExecutorAdded事件。CoarseG-rainedExecutorBackend收到RegisteredExecutor消息后，将会新建一个Executor执行器，并为此Executor充当信使与Driver通信。CoarseGrainedExecutorBackend收到RegisteredExecutor消息的源代码如下所示。

从上面代码中可以看到，CoarseGrainedExecutorBackend收到RegisteredExecutor（host-name）消息后，将会新建一个Executor。由此可见，Executor较CoarseGrainedExecutorBack-end后实例化，这与Executor和CoarseGrainedExecutorBackend的不同职责有关，Executor主要负责计算，而CoarseGrainedExecutorBackend主要负责通信，通信环境准备好了，架起了同CoarseGrainedSchedulerBackend通信的桥梁，就可以接收CoarseGrainedSchedulerBackend中调用launchTask方法发送的LaunchTask消息了，因此通信在前，计算在后。

Executor中的计算结果是通过CoarseGrainedExecutorBackend的statusUpdate方法返回给CoarseGrainedExecutorBackend的，statusUpdate方法的代码如下所示。

上面源代码中，通过参数taskId、state和data构建一个StatusUpdate对象，该对象将被当做消息发送到Driver端，Driver根据返回结果的需要，将会向CoarseGrainedExecutorBack-end发送新的指令消息，如LaunchTask、KillTask、StopExecutors和Shutdown等。