Spark源码分析之Executor分析

Executor是在worker启动的一个进程，用于执行task任务。我们知道CoarseGrainedSchedulerBackend在启动之后，会创建Driver终端，然后会立即向Driver发送RegisterExecutor消息，注册成功之后，会向CoarseGrainedSchedulerBackend返回一个RegisteredExecutor消息

然后会创建一个Executor对象；然后接下来Driver内的DAGScheduler

向CoarseGrainedSchedulerBackend发送LaunchTask消息，然后这个SchedulerBackend实际上调用executor来发起任务,创建TaskRunner来执行task

 

一 核心属性

String executorHostname: executor对应的hostname

Boolean isLocal:是否是本地的

ThreadPoolExecutor threadPool：线程池

boolean userClassPathFirst: 是否首先加载用户jar中的class

MutableURLClassLoader urlClassLoader: URL类加载器

Long maxDirectResultSize: 直接结果最大为多大，首先从spark.task.maxDirectResultSize获取，如果没有设置则，默认是1M,还需要和rpc的message所允许的最大字节数比较，看谁比较小

Long maxResultSize:最大结果限制，默认是1GB

ConcurrentHashMap runningTasks: 正在运行的task列表

ScheduledExecutorService heartbeater： 心跳线程

HEARTBEAT_MAX_FAILURES ：心跳检测失败默认值60

 

二 重要方法

2.1 launchTask

deflaunchTask(context:ExecutorBackend,
taskId: Long,
attemptNumber: Int,

    taskName: String,serializedTask:
ByteBuffer): Unit = {

  // 创建TaskRunner对象

  val tr=
new TaskRunner(context,taskId
= taskId,attemptNumber
= attemptNumber,
taskName,

    serializedTask)

  // <task id, task runner> 放入内存缓存中

  runningTasks.put(taskId,tr)

  // 调用TaskRunner的run方法

  threadPool.execute(tr)

}
 

2.2 killTask

def killTask(taskId: Long, interruptThread: Boolean): Unit = {
  // 获取task 对应的 TaskRunner
  val tr = runningTasks.get(taskId)
  // 调用kill方法kill task
  if (tr != null) {
    tr.kill(interruptThread)
  }
}

 

2.3 TaskRunner的run方法