解析Spark Executor内幕,详解CoarseGrainedExecutorBackend

超、凢脫俗 2022-06-10 10:50 282阅读 0赞

Spark Executor工作原理

ExecutorBackend注册

Executor实例化

Executor具体是如何工作的？

![Image 1][]

CoarseGrainedExecutorBackend，粗粒度的ExecutorBackend进程。

Worker为什么要启动另外一个进程？

Worker本身是管理当前机器上的资源，变动资源的时候向Master汇报。有很多应用程序，就需要很多Executor。这样程序之间不会一个奔溃导致所有的都奔溃。

1.在CoarseGrainedExecutorBackend启动时，向Driver注册Executor其实质是注册ExecutorBackend实例，和Executor实例之间没有直接的关系！！！

2.CoarseGrainedExecutorBackend是Executor运行所在的进程名称，Executor才是真正在处理Task的对象，Executor内部是通过线程池的方式来完成Task的计算的。

3. CoarseGrainedExecutorBackend和Executor是一一对应的。

4. CoarseGrainedExecutorBackend是一个消息通信体（其实现了ThreadSafeRpcEndpoint）。可以发送信息给Driver，并可以接收Driver中发过来的指令，例如启动Task等。

5.在Driver进程中，有两个至关重要的Endpoint，

a）第一个就是ClientEndpoint，主要负责向Master注册当前的程序；是AppClient的内部成员。

b）另外一个就是DriverEndpoint，这是整个程序运行时候的驱动器！！是CoarseGrainedExecutorBackend的内部成员。

6.在Driver中通过ExecutorData封装并注册ExecutorBackend的信息到Driver的内存数据结构ExecutorMapData中。ExecutorMapData是CoarseGrainedSchedulerBackend的成员。最终是注册给CoarseGrainedSchedulerBackend。

**private**\[cluster\] **class **ExecutorData(  
   **val **executorEndpoint: RpcEndpointRef,  
   **val **executorAddress: RpcAddress,  
   **override val **executorHost: String,  
   **var **freeCores: Int,  
   **override val **totalCores: Int,  
   **override val **logUrlMap: Map\[String, String\]  
) **extends **ExecutorInfo(executorHost, totalCores, logUrlMap)

7.实际在执行的时候，DriverEndpoint会把信息写入CoarseGrainedSchedulerBackend的内存数据结构ExecutorMapData中，所以说最终是注册给CoarseGrainedSchedulerBackend，也就是说CoarseGrainedSchedulerBackend掌握了为当前程序分配的所有的ExucutorBackend进程，而在每一个ExecutorBackend进程实例中会通过Executor对象来负责具体Task的运行。在欲行的时候使用syschronized关键字来保证ExecutorMapData安全的并发写操作。

8.CoarseGrainedExecutorBackend收到DriverEndpoint发送过来的RegisteredExecutor消息后会启动Executor实例对象，而Executor实例对象是事实上负责真正Task计算的。

Executor是如何工作的？

1.当Driver发送过来Task的时候，其实是发送给了CoarseGrainedExecutorBackend这个RpcEndpoint，而不是直接发送给了Executor（Executor由于不是消息循环体，所以永远无法直接接收远程发送过来的信息）。

case LaunchTask(data) =>
      if (executor == null) {
        logError("Received LaunchTask command but executor was null")
        System.exit(1)
      } else {
        val taskDesc = ser.deserialize[TaskDescription](data.value)
        logInfo("Got assigned task " + taskDesc.taskId)
        executor.launchTask(this, taskId = taskDesc.taskId, attemptNumber = taskDesc.attemptNumber,
          taskDesc.name, taskDesc.serializedTask)
      }

2.ExecutorBackend在收到Driver中发送过来的消息后，会通过调用launchTask来交给Executor去执行。

在介绍CoarseGrainedExecutorBackend的启动流程前，先了解下CoarseGrainedExecutorBackend。我们知道Executor负责计算任务，即执行task，而Executor对象的创建及维护是由CoarseGrainedExecutorBackend负责的，CoarseGrainedExecutorBackend在[Spark][]运行期是一个单独的进程，在Worker节点可以通过[Java][]的jps命令查看，如下   
![SouthEast][]   
如何启动上图中的CoarseGrainedExecutorBackend进程是我们后面介绍的重点，先看下其定义及UML   
![SouthEast 1][]   
![SouthEast 2][]   
从上图可以得到如下信息

1.  CoarseGrainedExecutorBackend是RpcEndpoint的子类，能够和Driver进行RPC通信，其生命周期方法onStart一定要关注，看执行了哪些动作。
2.  CoarseGrainedExecutorBackend维护了两个属性executor和driver，executor负责运行task，driver负责和Driver通信。
3.  ExecutorBackend有抽象方法statusUpdate，负责将Executor的计算结果返回给Driver。

最后，CoarseGrainedExecutorBackend是spark运行期的一个进程，Executor运行在该进程内。

# 启动CoarseGrainedExecutorBackend #

接着[Spark 任务调度之Launch Executor][Spark _Launch Executor]介绍，最后讲到Worker进程收到LaunchExecutor消息，然后，Worker将收到的消息封装为ExecutorRunner对象，调用其start方法，如下   
![SouthEast 3][]  
start方法启动线程，调用ExecutorRunner的fetchAndRunExecutor方法，fetchAndRunExecutor方法中将收到的信息拼接为[Linux][]命令，然后使用ProcessBuilder执行[linux][Linux]命令启动CoarseGrainedExecutorBackend，和启动Driver的方式如出一辙，fetchAndRunExecutor方法如下   
![SouthEast 4][]  
ProcessBuilder执行的Linux命令大致如下   
![SouthEast 5][]   
上面的java命令会调用CoarseGrainedExecutorBackend的main方法，main方法中处理命令行传入的参数，然后创建RpcEnv，并注册CoarseGrainedExecutorBackend，如下   
![SouthEast 6][]  
至此，启动CoarseGrainedExecutorBackend就介绍完了，由于将CoarseGrainedExecutorBackend注册到RpcEnv会调用其onStart方法，所以下一篇就是查看onStart方法中进行了哪些操作。

# 总结 #

介绍了CoarseGrainedExecutorBackend的作用及启动流程，大致如下   
![这里写图片描述][SouthEast 7]

[Image 1]: 
[Spark]: http://lib.csdn.net/base/spark
[Java]: http://lib.csdn.net/base/java
[SouthEast]: /images/20220610/aeff86f79d30410791f1a8cd2d8b9c1f.png
[SouthEast 1]: /images/20220610/b27e4f5f5d984acdbf7ad9689a16235d.png
[SouthEast 2]: /images/20220610/98090311ce8e44dd9392ff63c7898773.png
[Spark _Launch Executor]: http://blog.csdn.net/u011564172/article/details/69062351
[SouthEast 3]: /images/20220610/b3f144885bf34e13a64d5175ce261a22.png
[Linux]: http://lib.csdn.net/base/linux
[SouthEast 4]: /images/20220610/6b55a091534c432396f2b2f42725f076.png
[SouthEast 5]: /images/20220610/9255f7bfe29d4f9694e9aa738acb6652.png
[SouthEast 6]: /images/20220610/030e24cd508e42eeb4948c87be99ae22.png
[SouthEast 7]: /images/20220610/93387be79fd141f6bfff89b23c194e50.png