Spark 源码分析(九): Task 启动

前面已经分析到了 DAGScheduler 对 stage 划分,并对 Task 的最佳位置进行计算之后,通过调用 taskScheduler 的 submitTasks 方法,将每个 stage 的 taskSet 进行提交。

在 taskScheduler 的 submitTasks 方法中会为每个 taskSet 创建一个 TaskSetManager,用于管理 taskSet。然后向调度池中添加该 TaskSetManager,最后会调用 backend.reviveOffers() 方法为 task 分配资源。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
override def submitTasks(taskSet: TaskSet) {
//获取 taskSet 中的 task
val tasks = taskSet.tasks
logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
this.synchronized {
// 为每个 taskSet 创建一个 TaskSetManager
val manager = createTaskSetManager(taskSet, maxTaskFailures)
// 拿到 stage 的 id
val stage = taskSet.stageId
// 创建一个 HashMap ,用来存储 stage 对应的 TaskSetManager
val stageTaskSets =
taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
// 将上面创建的 taskSetManager 存入 map 中
stageTaskSets(taskSet.stageAttemptId) = manager
val conflictingTaskSet = stageTaskSets.exists { case (_, ts) =>
ts.taskSet != taskSet && !ts.isZombie
}
if (conflictingTaskSet) {
throw new IllegalStateException(s"more than one active taskSet for stage $stage:" +
s" ${stageTaskSets.toSeq.map{_._2.taskSet.id}.mkString(",")}")
}
// 向调度池中添加刚才创建的 TaskSetManager
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)

// 判断程序是否为 local 模式,并且 TaskSchedulerImpl 没有收到 Task
if (!isLocal && !hasReceivedTask) {
// 创建一个定时器,通过指定时间检查 TaskSchedulerImpl 的饥饿情况
starvationTimer.scheduleAtFixedRate(new TimerTask() {
override def run() {
// 如果 TaskSchedulerImpl 已经安排执行了 Task,则取消定时器
if (!hasLaunchedTask) {
logWarning("Initial job has not accepted any resources; " +
"check your cluster UI to ensure that workers are registered " +
"and have sufficient resources")
} else {
this.cancel()
}
}
}, STARVATION_TIMEOUT_MS, STARVATION_TIMEOUT_MS)
}
// 标记已经接收到 Task
hasReceivedTask = true
}
// 给 Task 分配资源
backend.reviveOffers()
}

下面主要看 backend.reviveOffers() 这个方法,在提交模式是 standalone 模式下,实际上是调用 StandaloneSchedulerBackend 的 reviveOffers 方法,实则调用的是其父类 CoarseGrainedSchedulerBackend 的 reviveOffers 方法,这个方法是向 driverEndpoint 发送一个 ReviveOffers 消息。

1
2
3
4
override def reviveOffers() {
// 向 driverEndpoint 发送 ReviveOffers 消息
driverEndpoint.send(ReviveOffers)
}

DriverEndpoint 收到信息后会调用 makeOffers 方法:

1
2
case ReviveOffers =>
makeOffers()

makeOffers 方法内部会将 application 所有可用的 executor 封装成一个 workOffers,每个 workOffers 内部封装了每个 executor 的资源数量。

然后调用 taskScheduler 的 resourceOffers 从上面封装的 workOffers 信息为每个 task 分配合适的 executor。

最后调用 launchTasks 启动 task。

1
2
3
4
5
6
7
8
9
10
private def makeOffers() {
// 过滤出可用的 executor
val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
// 将这些 executor 封装成 workOffers
val workOffers = activeExecutors.map { case (id, executorData) =>
new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
}.toIndexedSeq
// 给每个 task 分配 executor,然后调用 launchTasks 启动这些 task
launchTasks(scheduler.resourceOffers(workOffers))
}

下面看一下 launchTasks 这个方法。

这个方法主要做了这些操作:

1,遍历每个 task,然后将每个 task 信息序列化。

2,判断序列化后的 task 信息,如果大于 rpc 发送消息的最大值,则停止,建议调整 rpc 的 maxRpcMessageSize,如果小于 rpc 发送消息的最大值,则找到 task 对应的 executor,然后更新该 executor 对应的一些内存资源信息。

3,向 executor 发送 LaunchTask 消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
private def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
// 遍历所有的 task
for (task <- tasks.flatten) {
// 序列化 task 信息
val serializedTask = ser.serialize(task)
// 判断序列化后的 task 信息是否大于 rpc 能够传送的最大信息量
if (serializedTask.limit >= maxRpcMessageSize) {
scheduler.taskIdToTaskSetManager.get(task.taskId).foreach { taskSetMgr =>
try {
var msg = "Serialized task %s:%d was %d bytes, which exceeds max allowed: " +
"spark.rpc.message.maxSize (%d bytes). Consider increasing " +
"spark.rpc.message.maxSize or using broadcast variables for large values."
msg = msg.format(task.taskId, task.index, serializedTask.limit, maxRpcMessageSize)
taskSetMgr.abort(msg)
} catch {
case e: Exception => logError("Exception in error callback", e)
}
}
}
else {
// 找到对应的 executor
val executorData = executorDataMap(task.executorId)
// 更新 executor 的资源信息
executorData.freeCores -= scheduler.CPUS_PER_TASK

logDebug(s"Launching task ${task.taskId} on executor id: ${task.executorId} hostname: " +
s"${executorData.executorHost}.")

// 向 executor 发送 LaunchTask 消息
executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))
}
}
}

下面就看 Executor 收到消息后做了哪些操作,这里 executorData.executorEndpoint 实际上就是在创建 Executor 守护进程时候创建的那个 CoarseGrainedExecutorBackend。

所以找到 CoarseGrainedExecutorBackend 处理接收到 LaunchTask 消息后做了哪些操作。

首先他会判断当前的 executor 是不是为空,如果不为空就会反序列化 task 的信息,然后调用 executor 的 launchTask 方法。

1
2
3
4
5
6
7
8
9
10
11
12
case LaunchTask(data) =>
// 判断当前 executor 是不是空
if (executor == null) {
exitExecutor(1, "Received LaunchTask command but executor was null")
} else {
// 反序列化 task 的信息
val taskDesc = ser.deserialize[TaskDescription](data.value)
logInfo("Got assigned task " + taskDesc.taskId)
// 调用 executor 的 lauchTask 方法
executor.launchTask(this, taskId = taskDesc.taskId, attemptNumber = taskDesc.attemptNumber,
taskDesc.name, taskDesc.serializedTask)
}

下面就看下 executor 的 launchTask 做了哪些操作。

首先 executor 会为每个 task 创建一个 TaskRunner,然后会会将 task 添加到 runningTasks 的集合中,并标记其为运行状态,最后将 taskRunner 放到一个线程池中执行。

1
2
3
4
5
6
7
8
9
10
11
12
13
def launchTask(
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 放到线程池中执行
threadPool.execute(tr)
}

在 taskRunner 的 run 方法中回去执行每个 task,并会输出一系列的日志。task 运行完成后回向 driver 发送消息,driver 会更新 executor 的一些资源数据,并标记 task 已完成。

至此 task 启动完成。