spark的Executor是执行task的容器。和java的executor概念类似。
===================start executor runs task============================
->CoarseGrainedExecutorBackend::receiveWithLogging --接收CoarseGrainedSchedulerBackend发来的消息
->case LaunchTask(data) => 处理启动task的消息
->val taskDesc = ser.deserialize[TaskDescription](data.value) --将受到的taskDescription反序列化
->executor.launchTask(this, taskDesc.taskId, taskDesc.name, taskDesc.serializedTask) --调用executor的launchTask方法
->Executor::launchTask( --Executor执行task的方法
->val tr = new TaskRunner(context, taskId, taskName, serializedTask) --创建一个新的task，这样可以放到新的线程中执行
->override def run() {
->val (taskFiles, taskJars, taskBytes) = Task.deserializeWithDependencies(serializedTask) --解析task字段
->updateDependencies(taskFiles, taskJars) //Download any missing dependencies if we receive a new set of files
//and JARs from theSparkContext. Also adds any new JARs we fetched to the class loader. 更新并补全依赖
->for ((name, timestamp) <- newFiles if currentFiles.getOrElse(name, -1L) < timestamp) --获取依赖文件
->Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf, env.securityManager)
->for ((name, timestamp) <- newJars if currentJars.getOrElse(name, -1L) < timestamp) --获取依赖jar包
->Utils.fetchFile(name, new File(SparkFiles.getRootDirectory), conf, env.securityManager)
->val url = new File(SparkFiles.getRootDirectory, localName).toURI.toURL
->urlClassLoader.addURL(url)
->task = ser.deserialize[Task[Any]](taskBytes, Thread.currentThread.getContextClassLoader) --反序列化task
->val value = task.run(taskId.toInt) --直接调用task的run函数。
->val valueBytes = resultSer.serialize(value) --序列化任务结果
->val directResult = new DirectTaskResult(valueBytes, accumUpdates, task.metrics.orNull) 生成会送给的task结果
->val serializedDirectResult = ser.serialize(directResult) --序列化回送的结果
->execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult) --回送结果
->driver ! StatusUpdate(executorId, taskId, state, data) --CoarseGrainedSchedulerBackend实现
->env.shuffleMemoryManager.releaseMemoryForThisThread() // Release memory used by this thread for shuffles
->env.blockManager.memoryStore.releaseUnrollMemoryForThisThread()// Release memory used by this thread for unrolling blocks
->runningTasks.remove(taskId)
->runningTasks.put(taskId, tr)
->threadPool.execute(tr)
===========================end======================
/**
* Spark executor used with Mesos, YARN, and the standalone scheduler.
*/
private[spark] class Executor(
executorId: String,
slaveHostname: String,
properties: Seq[(String, String)],
isLocal: Boolean = false)
extends Logging
{
重要属性：
// Maintains the list of running tasks.
private val runningTasks = new ConcurrentHashMap[Long, TaskRunner]
通过心跳发送任务状态到master
def startDriverHeartbeater() {
终于看到熟悉的executor了。这就是最终我们要执行的东西。
def launchTask(
context: ExecutorBackend, taskId: Long, taskName: String, serializedTask: ByteBuffer) {
val tr = new TaskRunner(context, taskId, taskName, serializedTask)
runningTasks.put(taskId, tr)
threadPool.execute(tr)
}
class TaskRunner(
execBackend: ExecutorBackend, val taskId: Long, taskName: String, serializedTask: ByteBuffer)
CoarseGrainedExecutorBackend： 是直接与master的CoarseGrainedSchedulerBackend类对位的，它们直接通信来实现任务的传递和结果回送功能。
private[spark] class CoarseGrainedExecutorBackend(
driverUrl: String,
executorId: String,
hostPort: String,
cores: Int,
sparkProperties: Seq[(String, String)])
extends Actor with ActorLogReceive with ExecutorBackend with Logging {