Spark源码之TaskScheduler介绍篇
前面DAGScheduler将stage划分好之后,又将生成的TaskSet提交给TaskScheduler,那么本章节就要叙述下TaskScheduler如何启动Task的；
TaskScheduler任务源码分析
DAGScheduler将TaskSet提交给TaskScheduler,那么就先看下submitTasks(),打开TaskScheduler的实现类TaskSchedulerImpl,在这个方法里面，先生成了一个TaskManager对象来封装taskSet,然后判断当前stage中是否只正常运行一个taskSet，以及taskManager是否是僵尸进程；随后将生成的TaskManager放入到schedulableBuilder调度策略中，做完以上工作后开始想backend申请资源backend.reviveOffers();
override def submitTasks(taskSet: TaskSet) {
val tasks = taskSet.tasks
logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
this.synchronized {
//创建一个TaskManager maxTaskFailures:最大失败重试次数 默认4次
val manager = createTaskSetManager(taskSet, maxTaskFailures)
val stage = taskSet.stageId
val stageTaskSets =
taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
stageTaskSets(taskSet.stageAttemptId) = manager
// 在这里判断当前stage的是否有两个taskSet在运行，因为同一个stage中只能运行一个taskSet
// 一方面判断当前的TaskSet是否已经在运行了，
// 另一方面判断当前的taskSetManager是否是僵尸进程
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添加到调度策略中(FIFOSchedulableBuilder/FIARSchedulableBuilder)
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
if (!isLocal && !hasReceivedTask) {
starvationTimer.scheduleAtFixedRate(new TimerTask() {
override def run() {
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)
}
hasReceivedTask = true
}
//在这里开始向backend的发送资源申请的请求，其实是向DriverEndPoint发送的请求
backend.reviveOffers()
}
我们进入CoarseGrainedSchedulerBackend的reviveOffers方法，可以看到在方法里面driverEndpoint向自己发送了一个ReviveOffers消息，而这个driverEndpoint我们前面也讲过，就是当前应用程序的Driver;
override def reviveOffers() {
driverEndpoint.send(ReviveOffers)
}
进入driverEndpoint的reviveOffers方法，最终调用的是makeOffers()方法,在这个方法里面先过滤出状态为alive的executor，然后将这些activeExecutor封装成WorkerOffer对象，关键点是在最后的lanchTasks方法，我们先看下scheduler.resourceOffers(workOffers)这个方法的作用；
case ReviveOffers =>
makeOffers()
// Make fake resource offers on all executors
private def makeOffers() {
// Filter out executors under killing
val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
val workOffers = activeExecutors.map { case (id, executorData) =>
new WorkerOffer(id, executorData.executorHost, executorData.freeCores)
}.toSeq
launchTasks(scheduler.resourceOffers(workOffers))
}
我们再回到TaskSchedulerImpl,查看resourceOffers方法：
TaskSchedulerImpl.resourceOffers:为每一个Task具体分配计算资源，输入时ExecutorBackend以及可用的Cores，输出是 TaskDescription的二维数组，在其中确定了每个task具体运行在哪个ExecutorBackend；
在方法内部先将可用的executors添加到数据结构中，然后在将可用的executors进行shuffle以便做到负载均衡，为每个executor创建一个task数组用于存放TaskDescription，最后遍历调度策略中的TaskSet,使用就近原则为task分配executor，在这里需要腔调一点的是DAGScheduler.submitMissingTasks()方法中我们是获取了每个task的对应数据的位置，而在本方法中的taskSet.myLocalityLevels)是为了获取Task对应数据位置的级别,如下代码所示:
def resourceOffers(offers: Seq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {
// Mark each slave as alive and remember its hostname
// Also track if new executor is added
//todo 标记是否有新的executor
var newExecAvail = false
//todo 遍历每个executor
for (o <- offers) {
//todo 向数据结构中添加executor信息
executorIdToHost(o.executorId) = o.host
executorIdToTaskCount.getOrElseUpdate(o.executorId, 0)
if (!executorsByHost.contains(o.host)) {
executorsByHost(o.host) = new HashSet[String]()
executorAdded(o.executorId, o.host)
newExecAvail = true
}
for (rack <- getRackForHost(o.host)) {
hostsByRack.getOrElseUpdate(rack, new HashSet[String]()) += o.host
}
}
// Randomly shuffle offers to avoid always placing tasks on the same set of workers.
//todo 将可用的executor进行shuffle打乱,以便做到负载均衡
val shuffledOffers = Random.shuffle(offers)
// Build a list of tasks to assign to each worker.
//todo 为每个executor构建一个task数组
val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores))
//todo 所有executor可用的core资源
val availableCpus = shuffledOffers.map(o => o.cores).toArray
//todo 从调度池中获取TaskSetManagers
val sortedTaskSets = rootPool.getSortedTaskSetQueue
for (taskSet <- sortedTaskSets) {
logDebug("parentName: %s, name: %s, runningTasks: %s".format(
taskSet.parent.name, taskSet.name, taskSet.runningTasks))
if (newExecAvail) {
taskSet.executorAdded()
}
}
// Take each TaskSet in our scheduling order, and then offer it each node in increasing order
// of locality levels so that it gets a chance to launch local tasks on all of them.
// NOTE: the preferredLocality order: PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY
//todo 我们在DAGScheduler.submitMissingTasks()中已经获取了每个Task中数据所在的位置，
//todo 这是的taskSet.myLocalityLevels只是根据Task数据所在的host来获取它的的数据本地
//todo 性级别(PROCESS_LOCAL, NODE_LOCAL, NO_PREF, RACK_LOCAL, ANY)
var launchedTask = false
//todo 对每一个taskSet，按照就近顺序分配最近的executor来执行task
for (taskSet <- sortedTaskSets; maxLocality <- taskSet.myLocalityLevels) {
do {
//todo 将前面随机打散的WorkOffers计算资源按照就近原则分配给taskSet，用于执行其中的task
launchedTask = resourceOfferSingleTaskSet(
taskSet, maxLocality, shuffledOffers, availableCpus, tasks)
} while (launchedTask)
}
if (tasks.size > 0) {
hasLaunchedTask = true
}
return tasks
}
接着进入resourceOfferSingleTaskSet方法,遍历所有的executor的索引地址,以便作为tasks的索引，将每个task分配给相应的executor，并填充tasks；
而这个tasks数据结构是调用resourceOfferSingleTaskSet的方法里传进来的，它存储着每个executor内的tasks信息，详细信息见下图源码:
private def resourceOfferSingleTaskSet(
taskSet: TaskSetManager,
maxLocality: TaskLocality,
shuffledOffers: Seq[WorkerOffer],
availableCpus: Array[Int],
tasks: Seq[ArrayBuffer[TaskDescription]]) : Boolean = {
var launchedTask = false
//todo 遍历所有的executor的索引
for (i <- 0 until shuffledOffers.size) {
val execId = shuffledOffers(i).executorId
val host = shuffledOffers(i).host
//todo 判断该executor可以用的资源是否>=CPUS_PER_TASK(默认为1)
if (availableCpus(i) >= CPUS_PER_TASK) {
try {
//todo 为taskSet中的task分配executor，并将信息存储在tasks中,注意这个tasks是从
//todo 上面的方法传进来的
for (task <- taskSet.resourceOffer(execId, host, maxLocality)) {
//todo 将每个task信息写入下面的数据结构中
tasks(i) += task
val tid = task.taskId
taskIdToTaskSetManager(tid) = taskSet
taskIdToExecutorId(tid) = execId
executorIdToTaskCount(execId) += 1
executorsByHost(host) += execId
availableCpus(i) -= CPUS_PER_TASK
assert(availableCpus(i) >= 0)
launchedTask = true
}
} catch {
case e: TaskNotSerializableException =>
logError(s"Resource offer failed, task set ${taskSet.name} was not serializable")
// Do not offer resources for this task, but don't throw an error to allow other
// task sets to be submitted.
return launchedTask
}
}
}
return launchedTask
}
再到DriverEndPoint的makeOffers方法中，scheduler.resourceOffers(workOffers)已经执行完毕，taskSet已经分配完毕,接着执行launchTasks()方法，该方法遍历每个task,并向每个task所对应的executor发送launchTask消息;如下代码所示:
// Launch tasks returned by a set of resource offers
private def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
//todo 遍历每个task
for (task <- tasks.flatten) {
val serializedTask = ser.serialize(task)
//todo 检查序列化后的task大小
if (serializedTask.limit >= akkaFrameSize - AkkaUtils.reservedSizeBytes) {
scheduler.taskIdToTaskSetManager.get(task.taskId).foreach { taskSetMgr =>
try {
var msg = "Serialized task %s:%d was %d bytes, which exceeds max allowed: " +
"spark.akka.frameSize (%d bytes) - reserved (%d bytes). Consider increasing " +
"spark.akka.frameSize or using broadcast variables for large values."
msg = msg.format(task.taskId, task.index, serializedTask.limit, akkaFrameSize,
AkkaUtils.reservedSizeBytes)
taskSetMgr.abort(msg)
} catch {
case e: Exception => logError("Exception in error callback", e)
}
}
}
else {
//todo 获取task对应的executor
val executorData = executorDataMap(task.executorId)
executorData.freeCores -= scheduler.CPUS_PER_TASK
//todo 向该executor发送launchTask请求
executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))
}
}
}
我们继续追踪，进入CoarseGrainedExecutorBackend的LaunchTask的LaunchTask方法，该方法又调用了executor.launchTask；
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)
}
我们进入Executor的launchTask方法,在该方法内实例化除了TaskRunner对象,而TaskRunner对象是一个线程,通过线程池threadPool来运行;
def launchTask(
context: ExecutorBackend,
taskId: Long,
attemptNumber: Int,
taskName: String,
serializedTask: ByteBuffer): Unit = {
val tr = new TaskRunner(context, taskId = taskId, attemptNumber = attemptNumber, taskName,
serializedTask)
runningTasks.put(taskId, tr)
threadPool.execute(tr)
}
总结：
DAGScheduler划分好stage，并将生成的TaskSet提交给TaskScheduler，TaskScheduler向Driver请求分配资源，Driver将可用的ExecutorBackend资源发给TaskScheduler，在TaskScheduler中将Task分配给ExecutorBackend，最后向ExecutorBackend发送launchTask请求，在ExecutorBackend中调executor对象的launchTask，在Executor对象的launchTask方法中启动TaskRunner线程并用线程池去执行TaskRunner线程;
