首页 > 网络 > 云计算 >

Spark源码知识讲解之Executor分析

2017-11-11

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

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)
// 放入内存缓存中
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方法

相关文章
最新文章
热点推荐