Spark Executor原理

本文主要是介绍Spark Executor原理，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Master发指令给Worker启动Executor。

Worker接收到Master发送来的指令通过ExecutorRunner启动另外一个进程来启动Executor。

CoarseGrainedExecutorBackend通过发送RegisterExecutor向Driver注册，Driver在Executor注册成功后会返回RegisterExecutor信息给CoarseGrainedExecutorBackend。

一：Spark Executor工作原理 
1. 需要特别注意是在CoarseGrainedExecutorBackend启动时向Driver注册Executor其实质是注册ExecutorBackend实例，和Executor无关。 
2．CoarseGrainedExecutorBackend是Executor运行所在的进程名称，CoarseGrainedExecutorBackend本身并不会完成具体任务的计算，Executor才会完成计算。Executor才是处理Task的对象，Executor内部是通过线程池的方式来完成Task的计算的； 
3．CoarseGrainedExecutorBackend是一个进程，里面有一个Executor对象，CoarseGrainedExecutorBackend和Executor是一一对应的； 
4．CoarseGrainedExecutorBackend是一个消息通信体(其实现了ThreadSafeRpcEndPoint)。可以发送消息给Driver并可以接受Driver中发过来的指令，例如启动Task等； 
5. 在Driver进程有两个至关重要的Endpoint: 
a) ClientEndpoint: 主要负责向Master注册当前的程序,是AppClient的内部成员； 
b) DriverEndpoint:这是整个程序运行时候的驱动器,例如接收CoarseGrainedExecutorBackend的注册，是CoarseGrainedExecutorBackend的内部成员。

Executor内幕原理和运行流程解密

1. CoarseGrainedExecutorBackend通过onStart()发送RegisterExecutor向Driver注册。这里面的ref也就相当于Driver。

override def onStart() {logInfo("Connecting to driver: " + driverUrl)rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref =>// This is a very fast action so we can use "ThreadUtils.sameThread"driver = Some(ref)ref.ask[RegisterExecutorResponse](RegisterExecutor(executorId, self, hostPort, cores, extractLogUrls))
1
2
3
4
5
6
7
1
2
3
4
5
6
7

RegisterExecutor此时的命名是有问题的，因为向Driver端注册并不是注册Executor，而是注册ExecutorBackend. 
2. Driver要接受ExecutorBackend的注册，也就是接受CoarseGrainedExecutorBackend发来的请求。在DriverEndpoint中会接收到RegisterExecutor信息并完成在Driver的注册。

override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {case RegisterExecutor(executorId, executorRef, hostPort, cores, logUrls) =>if (executorDataMap.contains(executorId)) {context.reply(RegisterExecutorFailed("Duplicate executor ID: " + executorId))
1
2
3
4
5
1
2
3
4
5

接收到请求之后Driver是怎么办的？ 
在Driver中通过ExecutorData封装并注册ExecutorBackend的信息到Driver的内存数据结构executorMapData中；

private val executorDataMap = new HashMap[String, ExecutorData]
1
1

为啥是ExecutorBackend？ 
因为CoarseGrainedExecutorBackend是继承ThreadSafeRpcEndpoint，而Executor没有继承ThreadSafeRpcEndpo

这篇关于Spark Executor原理的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---