Spark Executor内幕

Spark Executor工作原理

ExecutorBackend注册

Executor实例化

Executor具体是如何工作的？

 

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去执行。

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)
  }