Spark作业执行流程源码解析
本文梳理一下Spark作业执行的流程。
本文梳理一下Spark作业执行的流程。
Spark作业和任务调度系统是其核心,通过内部RDD的依赖DAG,使得模块之间的调用和处理变得游刃有余。
相关概念
Job(作业):通过行动操作生成的一个或多个调度阶段
Stage:根据依赖关系划分的多个任务集,称为调度阶段,也叫做TaskSet(任务集)。划分Stage是由DAGScheduler进行的,任务阶段分为Shuffle Map Stage和Result Stage。
Task:是Spark执行计算的最小单位,会被分发到Executor中执行。
DAGScheduler:是面向调度阶段的任务调度器,接收Spark应用提交的作业,根据依赖关系划分stage,并提交给TaskScheduler。
TaskScheduler:是面向任务的 调度器,接收DAGScheduler划分好的stage,发送给Worker节点的Executor运行任务。
关于RDD相关知识、行动操作、宽窄依赖请参考Spark RDD基本概念、宽窄依赖、转换行为操作
概述
Spark作业主要是根据我们编写的业务处理代码,生成一系列相互依赖的调度阶段,之后将调度阶段中的任务提交Executor的执行的过程。
上图是spark作业运行流程图。主要分为四块:
-
构建DAG
行动操作触发提交作业,提交之后根据依赖关系构造DAG。
-
划分调度阶段、提交调度阶段
DAGScheduler中根据宽依赖划分调度阶段(stage)。每个stage包含多个task,组成taskset提交给TaskScheduler执行
-
通过集群管理器启动任务
TaskScheduler收到DAGScheduler提交的任务集,以任务的形式一个个分发到Executor中进行执行。
-
Executor端执行任务,完成后存储报告结果
Executor接到任务后,扔到线程池中执行任务。任务完成后,报告结果给Driver。
源码解析
从以下的代码展开叙述:
def main(args: Array[String]): Unit = {
val sc = new SparkContext("local", "word-count", new SparkConf())
val words = Seq("hello spark", "hello scala", "hello java")
val rdd = sc.makeRDD(words)
rdd
.flatMap(_.split(" "))
.map((_, 1))
.reduceByKey(_ + _)
.sortByKey()
.foreach(println(_))
}
这是一个简单的WordCount案例。首先根据序列生成RDD,再经过一系列算子调用计算word的个数,之后再进行排序,输出结果。
作业提交
上面的代码中,flatMap、map、reduceByKey、sortByKey都是转化算子,不会触发计算;foreach是行动算子,会提交作业,触发计算。
看看foreach的内部的实现:
def foreach(f: T => Unit): Unit = withScope {
val cleanF = sc.clean(f)
// 将当前rdd引用和我们编写的函数传给sc.runJob
sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))
}
// 以下runJob函数都是SparkContext内部的重载函数
def runJob[T, U: ClassTag](rdd: RDD[T], func: Iterator[T] => U): Array[U] = {
// 添加分区信息
runJob(rdd, func, 0 until rdd.partitions.length)
}
def runJob[T, U: ClassTag](
rdd: RDD[T],
func: Iterator[T] => U,
partitions: Seq[Int]): Array[U] = {
val cleanedFunc = clean(func)
runJob(rdd, (ctx: TaskContext, it: Iterator[T]) => cleanedFunc(it), partitions)
}
def runJob[T, U: ClassTag](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int]): Array[U] = {
// 创建一个数组来保存结果
val results = new Array[U](partitions.size)
runJob[T, U](rdd, func, partitions, (index, res) => results(index) = res)
results
}
// 多次调用runJob,之后将调用DAGScheduler的runJob提交作业
def runJob[T, U: ClassTag](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
// 任务成功后的处理函数
resultHandler: (Int, U) => Unit): Unit = {
if (stopped.get()) {
throw new IllegalStateException("SparkContext has been shutdown")
}
val callSite = getCallSite
val cleanedFunc = clean(func)
logInfo("Starting job: " + callSite.shortForm)
if (conf.getBoolean("spark.logLineage", false)) {
logInfo("RDD\'s recursive dependencies:\n" + rdd.toDebugString)
}
// 调用DAGScheduler.runJob提交作业
dagScheduler.runJob(rdd, cleanedFunc, partitions, callSite, resultHandler, localProperties.get)
progressBar.foreach(_.finishAll())
rdd.doCheckpoint()
}
foreach内部调用了SparkContext.runJob()提交作业,SparkContext内部反复调用了几次重载的runJob方法。
runJob最终的参数中有当前rdd的引用、处理逻辑函数、分区数等,之后调用DagScheduler.runJob()提交作业。
现在再来到DagScheduler.runJob(),看看内部调用:
def runJob[T, U](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: CallSite,
resultHandler: (Int, U) => Unit,
properties: Properties): Unit = {
val start = System.nanoTime
// 提交作业
// waiter是等待DAGScheduler作业完成的对象。
// 任务完成后,它将结果传递给给定的处理函数
val waiter = submitJob(rdd, func, partitions, callSite, resultHandler, properties)
ThreadUtils.awaitReady(waiter.completionFuture, Duration.Inf)
waiter.completionFuture.value.get match {
case scala.util.Success(_) =>
case scala.util.Failure(exception) =>
val callerStackTrace = Thread.currentThread().getStackTrace.tail
exception.setStackTrace(exception.getStackTrace ++ callerStackTrace)
throw exception
}
}
// 提交job,划分调度阶段
def submitJob[T, U](
rdd: RDD[T],
func: (TaskContext, Iterator[T]) => U,
partitions: Seq[Int],
callSite: CallSite,
resultHandler: (Int, U) => Unit,
properties: Properties): JobWaiter[U] = {
// 检查以确保我们没有在不存在的分区上启动任务。
val maxPartitions = rdd.partitions.length
partitions.find(p => p >= maxPartitions || p < 0).foreach { p =>
throw new IllegalArgumentException(
"Attempting to access a non-existent partition: " + p + ". " +
"Total number of partitions: " + maxPartitions)
}
// 为当前job获取id
val jobId = nextJobId.getAndIncrement()
// 如果分区为0,返回一个空job
if (partitions.size == 0) {
return new JobWaiter[U](this, jobId, 0, resultHandler)
}
assert(partitions.size > 0)
val func2 = func.asInstanceOf[(TaskContext, Iterator[_]) => _]
// 封装waiter,用于在执行结束时,回调处理结果
val waiter = new JobWaiter(this, jobId, partitions.size, resultHandler)
// eventProcessLoop是用于提交/接收DAG调度事件的事件环
// 提交作业,告知DAGScheduler开始划分调度阶段。
eventProcessLoop.post(JobSubmitted(
jobId, rdd, func2, partitions.toArray, callSite, waiter,
SerializationUtils.clone(properties)))
waiter
}
内部调用了submitJob(),发送提交作业的消息到DAGScheduler的eventProcessLoop事件环中。
划分&提交调度阶段
eventProcessLoop是用于接收调度事件的调度环,对应的类是DAGSchedulerEventProcessLoop。
内部通过模式匹配接收消息,作出相应处理。接收到提交作业的消息后,调用dagScheduler.handleJobSubmitted()开始划分调度阶段、提交调度阶段。
private def doOnReceive(event: DAGSchedulerEvent): Unit = event match {
// 匹配提交作业的消息
case JobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties) =>
dagScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)
}
看看dagScheduler.handleJobSubmitted()内部:
private[scheduler] def handleJobSubmitted(jobId: Int,
finalRDD: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
callSite: CallSite,
listener: JobListener,
properties: Properties) {
var finalStage: ResultStage = null
try {
// 根据依赖关系创建ResultStage
finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite)
} catch {
...
}
// 提交作业,清除内部数据
barrierJobIdToNumTasksCheckFailures.remove(jobId)
// 通过jobId, finalStage创建job
val job = new ActiveJob(jobId, finalStage, callSite, listener, properties)
val jobSubmissionTime = clock.getTimeMillis()
// 将job存入jobId映射到job的map中
jobIdToActiveJob(jobId) = job
activeJobs += job
finalStage.setActiveJob(job)
val stageIds = jobIdToStageIds(jobId).toArray
val stageInfos = stageIds.flatMap(id => stageIdToStage.get(id).map(_.latestInfo))
listenerBus.post(
SparkListenerJobStart(job.jobId, jobSubmissionTime, stageInfos, properties))
// 提交调度阶段
submitStage(finalStage)
}
handleJobSubmitted主要分为两块,一块是根据依赖生成ResultStage,一块是提交ResultStage。
生成ResultStage
先看一下生成ResultStage,也就是createResultStage方法。
private def createResultStage(
rdd: RDD[_],
func: (TaskContext, Iterator[_]) => _,
partitions: Array[Int],
jobId: Int,
callSite: CallSite): ResultStage = {
checkBarrierStageWithDynamicAllocation(rdd)
checkBarrierStageWithNumSlots(rdd)
checkBarrierStageWithRDDChainPattern(rdd, partitions.toSet.size)
// 先获取当前rdd的父调度阶段
val parents = getOrCreateParentStages(rdd, jobId)
val id = nextStageId.getAndIncrement()
val stage = new ResultStage(id, rdd, func, partitions, parents, jobId, callSite)
stageIdToStage(id) = stage
updateJobIdStageIdMaps(jobId, stage)
stage
}
会首先获取当前RDD的父阶段,获取后根据父阶段,创建ResultStage。
这里注意一下,这里的rdd是ShuffledRDD的引用。因为我们foreach触发计算的时候,将调用rdd的引用传了进来,也就是sortByKey生成的ShuffledRDD的引用。
接着看getOrCreateParentStages()是怎么获取当前RDD的父阶段的:
private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
// 获取宽依赖,之后根据获取的宽依赖,创建对应的ShuffleMapStage
getShuffleDependencies(rdd).map { shuffleDep =>
getOrCreateShuffleMapStage(shuffleDep, firstJobId)
}.toList
}
// 获取当前RDD的宽依赖
// 返回作为给定RDD的直接父级的shuffle依赖项
// 此函数将不会返回更远的祖先。例如,如果C对B具有宽依赖性,而B对A具有宽依赖性
// A <-- B <-- C
// 用rdd C调用此函数只会返回B <-C依赖项。
private[scheduler] def getShuffleDependencies(
rdd: RDD[_]): HashSet[ShuffleDependency[_, _, _]] = {
val parents = new HashSet[ShuffleDependency[_, _, _]]
val visited = new HashSet[RDD[_]]
val waitingForVisit = new ArrayStack[RDD[_]]
waitingForVisit.push(rdd)
while (waitingForVisit.nonEmpty) {
val toVisit = waitingForVisit.pop()
if (!visited(toVisit)) {
visited += toVisit
toVisit.dependencies.foreach {
case shuffleDep: ShuffleDependency[_, _, _] =>
parents += shuffleDep
case dependency =>
waitingForVisit.push(dependency.rdd)
}
}
}
parents
}
// 如果shuffle map stage已在shuffleIdToMapStage中存在,则获取
// 不存在的话,将创建shuffle map stage
private def getOrCreateShuffleMapStage(
shuffleDep: ShuffleDependency[_, _, _],
firstJobId: Int): ShuffleMapStage = {
shuffleIdToMapStage.get(shuffleDep.shuffleId) match {
case Some(stage) =>
stage
case None =>
// 查找尚未在shuffleToMapStage中注册的祖先shuffle依赖项,
// 并为它创建shuffle map stage
getMissingAncestorShuffleDependencies(shuffleDep.rdd).foreach { dep =>
if (!shuffleIdToMapStage.contains(dep.shuffleId)) {
createShuffleMapStage(dep, firstJobId)
}
}
// 为当前shuffle依赖创建shuffle map stage
createShuffleMapStage(shuffleDep, firstJobId)
}
}
getOrCreateParentStages中先调用getShuffleDependencies(),获取当前RDD的宽依赖;获取后,调用getOrCreateShuffleMapStage()为宽依赖创建stage(如果stage已存在就直接获取)。
先说一下getShuffleDependencies方法,如代码注释所说:返回作为给定RDD的直接父级的shuffle依赖项,不会返回整个DAG上所有的宽依赖。另外说一下,getShuffleDependencies这种写法感觉极度舒适,之后还有一个方法也是这么写。
我们传入的RDD是sortByKey生成的ShuffleRDD实例,调用getShuffleDependencies就会返回ShuffleDependency。
再说一下getOrCreateShuffleMapStage方法,它为返回的ShuffleDependency创建shuffle map stage。
它内部会在shuffleIdToMapStage中找当前ShuffleDependency是否存在stage,如果存在则返回,不存在则创建。
在创建之前,首先会调用getMissingAncestorShuffleDependencies()获取当前依赖的所有祖先宽依赖,并判断他们是否存在对应的调度阶段,如果不存在则调用createShuffleMapStage()创建。确保所有祖先宽依赖都存在对应的调度阶段后,调用createShuffleMapStage()为当前ShuffleDependency创建stage。
看看getMissingAncestorShuffleDependencies和createShuffleMapStage的实现:
// 查找所有尚未在shuffleToMapStage中注册的祖先shuffle依赖项
private def getMissingAncestorShuffleDependencies(
rdd: RDD[_]): ArrayStack[ShuffleDependency[_, _, _]] = {
val ancestors = new ArrayStack[ShuffleDependency[_, _, _]]
val visited = new HashSet[RDD[_]]
val waitingForVisit = new ArrayStack[RDD[_]]
waitingForVisit.push(rdd)
while (waitingForVisit.nonEmpty) {
val toVisit = waitingForVisit.pop()
if (!visited(toVisit)) {
visited += toVisit
// 获取宽依赖
getShuffleDependencies(toVisit).foreach { shuffleDep =>
if (!shuffleIdToMapStage.contains(shuffleDep.shuffleId)) {
ancestors.push(shuffleDep)
waitingForVisit.push(shuffleDep.rdd)
}
}
}
}
ancestors
}
// 为shuffle依赖创建shuffle map stage
def createShuffleMapStage(shuffleDep: ShuffleDependency[_, _, _], jobId: Int): ShuffleMapStage = {
val rdd = shuffleDep.rdd
checkBarrierStageWithDynamicAllocation(rdd)
checkBarrierStageWithNumSlots(rdd)
checkBarrierStageWithRDDChainPattern(rdd, rdd.getNumPartitions)
val numTasks = rdd.partitions.length
val parents = getOrCreateParentStages(rdd, jobId)
val id = nextStageId.getAndIncrement()
val stage = new ShuffleMapStage(
id, rdd, numTasks, parents, jobId, rdd.creationSite, shuffleDep, mapOutputTracker)
stageIdToStage(id) = stage
// 创建stage时会将stage放入shuffleId映射到stage的Map中
shuffleIdToMapStage(shuffleDep.shuffleId) = stage
updateJobIdStageIdMaps(jobId, stage)
if (!mapOutputTracker.containsShuffle(shuffleDep.shuffleId)) {
mapOutputTracker.registerShuffle(shuffleDep.shuffleId, rdd.partitions.length)
}
stage
}
getMissingAncestorShuffleDependencies和getShuffleDependencies的实现方法类似,返回所有尚未在shuffleToMapStage中注册的祖先shuffle依赖项。createShuffleMapStage为shuffle dependency创建shuffle map stage。
到此,getOrCreateParentStages的步骤就走完了,也就获取到了当前rdd的父阶段。
视线回到createResultStage方法中来:
val stage = new ResultStage(id, rdd, func, partitions, parents, jobId, callSite)
将stageId、rdd、处理逻辑方法、分区、父调度阶段等作为参数构造ResultStage。ResultStage就生成成功了。
提交ResultStage
在handleJobSubmitted方法中,调用submitStage()将生成的ResultStage提交。
看看submitStage内部:
// 提交阶段,但首先递归提交所有丢失的父阶段
private def submitStage(stage: Stage) {
val jobId = activeJobForStage(stage)
if (jobId.isDefined) {
logDebug("submitStage(" + stage + ")")
// 如果当前阶段不是在等待&不是在运行&没有结束,开始运行
if (!waitingStages(stage) && !runningStages(stage) && !failedStages(stage)) {
val missing = getMissingParentStages(stage).sortBy(_.id)
logDebug("missing: " + missing)
if (missing.isEmpty) {
logInfo("Submitting " + stage + " (" + stage.rdd + "), which has no missing parents")
submitMissingTasks(stage, jobId.get)
} else {
for (parent <- missing) {
submitStage(parent)
}
waitingStages += stage
}
}
} else {
abortStage(stage, "No active job for stage " + stage.id, None)
}
}
submitStage先调用getMissingParentStages获取所有丢失的父阶段。
如果没有丢失的父阶段,才会调用submitMissingTasks()提交当前阶段的任务集;如果存在丢失的父阶段,则递归调用submitStage先提交父阶段。
getMissingParentStages的实现方式和getShuffleDependencies也类似,这里就不看了,它的作用就是获取所有丢失的父阶段。
再大致说一下submitMissingTasks()是怎么提交任务的:
val tasks: Seq[Task[_]] = try {
stage match {
case stage: ShuffleMapStage =>
stage.pendingPartitions.clear()
partitionsToCompute.map { id =>
val locs = taskIdToLocations(id)
val part = partitions(id)
stage.pendingPartitions += id
// 创建shuffleMapTask
new ShuffleMapTask(stage.id, stage.latestInfo.attemptNumber,
taskBinary, part, locs, properties, serializedTaskMetrics, Option(jobId),
Option(sc.applicationId), sc.applicationAttemptId, stage.rdd.isBarrier())
}
case stage: ResultStage =>
partitionsToCompute.map { id =>
val p: Int = stage.partitions(id)
val part = partitions(p)
val locs = taskIdToLocations(id)
// 创建ResultTask
new ResultTask(stage.id, stage.latestInfo.attemptNumber,
taskBinary, part, locs, id, properties, serializedTaskMetrics,
Option(jobId), Option(sc.applicationId), sc.applicationAttemptId,
stage.rdd.isBarrier())
}
}
}
if (tasks.size > 0) {
// 调用taskScheduler.submitTasks()提交task
taskScheduler.submitTasks(new TaskSet(
tasks.toArray, stage.id, stage.latestInfo.attemptNumber, jobId, properties))
}
submitMissingTasks内部根据ShuffleMapStage和ResultStage分别生成ShuffleMapTask和ResultTask。
之后将task封装为TaskSet,调用TaskScheduler.submitTasks()提交任务。
到这里,划分和提交调度阶段已经走完了。接下来开始看提交任务的源码。
提交任务
上面调用了TaskScheduler.submitTasks()提交任务,TaskScheduler是特质,真正方法实现在类TaskSchedulerImpl中,我们看看内部实现:
override def submitTasks(taskSet: TaskSet) {
val tasks = taskSet.tasks
this.synchronized {
// 为该TaskSet创建TaskSetManager,管理这个任务集的生命周期
val manager = createTaskSetManager(taskSet, maxTaskFailures)
val stage = taskSet.stageId
val stageTaskSets =
taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
stageTaskSets.foreach { case (_, ts) =>
ts.isZombie = true
}
stageTaskSets(taskSet.stageAttemptId) = manager
// 将该任务集的管理器加入到系统调度池中去,由系统统一调度
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.reviveOffers()
}
会首先为每个TaskSet创建TaskSetManager用于管理整个TaskSet的生命周期,并调用schedulableBuilder.addTaskSetManager
将任务集管理器添加到系统调度池中去,之后调用SchedulerBackend.reviveOffers()分配资源并运行
看一下SchedulerBackend的其中一个子类CoarseGrainedSchedulerBackend的实现:
override def reviveOffers() {
// 向Driver发送ReviveOffsers的消息
driverEndpoint.send(ReviveOffers)
}
内部会向Driver终端点发送ReviveOffers的消息,分配资源并运行。
CoarseGrainedSchedulerBackend的实例就是代表Driver端的守护进程,其实也相当于自己发给自己。
接收到ReviveOffers的消息后,会调用makeOffers()。
看看makeOffers()实现:
private def makeOffers() {
val taskDescs = withLock {
// 获取集群中可用的Executor列表
val activeExecutors = executorDataMap.filterKeys(executorIsAlive)
val workOffers = activeExecutors.map {
case (id, executorData) =>
new WorkerOffer(id, executorData.executorHost, executorData.freeCores,
Some(executorData.executorAddress.hostPort))
}.toIndexedSeq
// 分配运行资源
scheduler.resourceOffers(workOffers)
}
if (!taskDescs.isEmpty) {
// 提交任务
launchTasks(taskDescs)
}
}
makeOffers()内部会先获取所有可用的Executor列表,然后调用TaskSchedulerImpl.resourceOffers()分配资源,分配资源完成后,调用launchTask()提交任务。
看看TaskSchedulerImpl.resourceOffers()的实现:
// 由集群管理器调用以在slave上提供资源。
def resourceOffers(offers: IndexedSeq[WorkerOffer]): Seq[Seq[TaskDescription]] = synchronized {
//将每个slave标记为活动并记住其主机名, 还跟踪是否添加了新的Executor
var newExecAvail = false
for (o <- offers) {
if (!hostToExecutors.contains(o.host)) {
hostToExecutors(o.host) = new HashSet[String]()
}
if (!executorIdToRunningTaskIds.contains(o.executorId)) {
hostToExecutors(o.host) += o.executorId
executorAdded(o.executorId, o.host)
executorIdToHost(o.executorId) = o.host
executorIdToRunningTaskIds(o.executorId) = HashSet[Long]()
newExecAvail = true
}
for (rack <- getRackForHost(o.host)) {
hostsByRack.getOrElseUpdate(rack, new HashSet[String]()) += o.host
}
}
// 移除黑名单中的节点
blacklistTrackerOpt.foreach(_.applyBlacklistTimeout())
val filteredOffers = blacklistTrackerOpt.map { blacklistTracker =>
offers.filter { offer =>
!blacklistTracker.isNodeBlacklisted(offer.host) &&
!blacklistTracker.isExecutorBlacklisted(offer.executorId)
}
}.getOrElse(offers)
// 为任务随机分配Executor,避免任务集中分配到Worker上
val shuffledOffers = shuffleOffers(filteredOffers)
// 存储已分配好的任务
val tasks = shuffledOffers.map(o => new ArrayBuffer[TaskDescription](o.cores / CPUS_PER_TASK))
val availableCpus = shuffledOffers.map(o => o.cores).toArray
val availableSlots = shuffledOffers.map(o => o.cores / CPUS_PER_TASK).sum
// 获取按照调度策略排序好的TaskSetManager
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()
}
}
// 为排好序的TaskSetManager列表进行分配资源。分配的原则是就近原则,按照顺序为PROCESS_LOCAL、NODE_LOCAL、NO_PREF、RACK_LOCAL、ANY
for (taskSet <- sortedTaskSets) {
if (taskSet.isBarrier && availableSlots < taskSet.numTasks) {
...
} else {
var launchedAnyTask = false
val addressesWithDescs = ArrayBuffer[(String, TaskDescription)]()
for (currentMaxLocality <- taskSet.myLocalityLevels) {
var launchedTaskAtCurrentMaxLocality = false
do {
launchedTaskAtCurrentMaxLocality = resourceOfferSingleTaskSet(taskSet,
currentMaxLocality, shuffledOffers, availableCpus, tasks, addressesWithDescs)
launchedAnyTask |= launchedTaskAtCurrentMaxLocality
} while (launchedTaskAtCurrentMaxLocality)
}
...
}
}
if (tasks.size > 0) {
hasLaunchedTask = true
}
return tasks
}
resourceOffers中按照调度策略、就近原则为Task分配资源,返回分配好资源的Task。
分配好资源后,调用launchTasks()提交任务。
private def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
for (task <- tasks.flatten) {
// 序列化任务
val serializedTask = ser.serialize(task)
if (serializedTask.limit >= maxRpcMessageSize) {
...
}
else {
val executorData = executorDataMap(task.executorId)
executorData.freeCores -= scheduler.CPUS_PER_TASK
// 向Executor所在节点终端发送LaunchTask的消息
executorData.executorEndpoint.send(LaunchTask(new SerializableBuffer(serializedTask)))
}
}
}
launchTasks内部先将任务序列化,之后把任务一个个的发送到对应的CoarseGrainedExecutorBackend进行执行。
至此任务就提交完成了,接下来看Executor是如何执行任务的。
执行任务
CoarseGrainedExecutorBackend接收到LaunchTask消息后,会调用Executor.launchTask()执行任务。
override def receive: PartialFunction[Any, Unit] = {
case LaunchTask(data) =>
if (executor == null) {
exitExecutor(1, "Received LaunchTask command but executor was null")
} else {
val taskDesc = TaskDescription.decode(data.value)
logInfo("Got assigned task " + taskDesc.taskId)
// 调用Executor.launchTask执行任务
executor.launchTask(this, taskDesc)
}
}
看看Executor.launchTask的实现:
def launchTask(context: ExecutorBackend, taskDescription: TaskDescription): Unit = {
// 将Task封装到TaskRunner中
val tr = new TaskRunner(context, taskDescription)
runningTasks.put(taskDescription.taskId, tr)
// 将TaskRunner扔到线程池中进行执行
threadPool.execute(tr)
}
launchTask中会将Task封装到TaskRunner中,然后把TaskRunner扔到线程池中进行执行。
TaskRunner是一个线程类,看一下它run方法的操作:
override def run(): Unit = {
threadId = Thread.currentThread.getId
Thread.currentThread.setName(threadName)
val threadMXBean = ManagementFactory.getThreadMXBean
val taskMemoryManager = new TaskMemoryManager(env.memoryManager, taskId)
val deserializeStartTime = System.currentTimeMillis()
val deserializeStartCpuTime = if (threadMXBean.isCurrentThreadCpuTimeSupported) {
threadMXBean.getCurrentThreadCpuTime
} else 0L
Thread.currentThread.setContextClassLoader(replClassLoader)
val ser = env.closureSerializer.newInstance()
// 开始运行
execBackend.statusUpdate(taskId, TaskState.RUNNING, EMPTY_BYTE_BUFFER)
var taskStartTime: Long = 0
var taskStartCpu: Long = 0
startGCTime = computeTotalGcTime()
try {
// 反序列化任务
task = ser.deserialize[Task[Any]](
taskDescription.serializedTask, Thread.currentThread.getContextClassLoader)
// value是返回结果
val value = Utils.tryWithSafeFinally {
// 调用Task.run运行Task,并获取返回结果
val res = task.run(
taskAttemptId = taskId,
attemptNumber = taskDescription.attemptNumber,
metricsSystem = env.metricsSystem)
threwException = false
res
} {
val releasedLocks = env.blockManager.releaseAllLocksForTask(taskId)
val freedMemory = taskMemoryManager.cleanUpAllAllocatedMemory()
}
val resultSer = env.serializer.newInstance()
val beforeSerialization = System.currentTimeMillis()
val valueBytes = resultSer.serialize(value)
val afterSerialization = System.currentTimeMillis()
val directResult = new DirectTaskResult(valueBytes, accumUpdates)
val serializedDirectResult = ser.serialize(directResult)
val resultSize = serializedDirectResult.limit()
// 执行结果的处理
val serializedResult: ByteBuffer = {
// 结果大于maxResultSize,直接丢弃;这个值通过spark.driver.maxResultSize进行设置
if (maxResultSize > 0 && resultSize > maxResultSize) {
ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
}
// 结果大于maxDirectResultSize,存放到BlockManager中,然后将BlockId发送到Driver
else if (resultSize > maxDirectResultSize) {
val blockId = TaskResultBlockId(taskId)
env.blockManager.putBytes(
blockId,
new ChunkedByteBuffer(serializedDirectResult.duplicate()),
StorageLevel.MEMORY_AND_DISK_SER)
ser.serialize(new IndirectTaskResult[Any](blockId, resultSize))
}
// 直接将结果发到Driver
else {
serializedDirectResult
}
}
// 任务执行完成,调用CoarseGrainedExecutorBackend.statusUpdate
execBackend.statusUpdate(taskId, TaskState.FINISHED, serializedResult)
} catch {
...
} finally {
runningTasks.remove(taskId)
}
}
run方法中,会将任务反序列化,然后调用Task.run()执行Task;执行完成后获取执行结果,根据结果的大小分情况处理,之后调用CoarseGrainedExecutorBackend.statusUpdate()向Driver汇报执行结果。
Task的run方法中,会调用runTask()执行任务。
Task是抽象类,没有对runTask()进行实现。具体的实现是由ShuffleMapTask和ResultTask进行的。
先看看ShuffleMapTask的runTask的实现:
override def runTask(context: TaskContext): MapStatus = {
val threadMXBean = ManagementFactory.getThreadMXBean
val deserializeStartTime = System.currentTimeMillis()
val deserializeStartCpuTime = if (threadMXBean.isCurrentThreadCpuTimeSupported) {
threadMXBean.getCurrentThreadCpuTime
} else 0L
// 反序列化
val ser = SparkEnv.get.closureSerializer.newInstance()
val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])](
ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)
_executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime
_executorDeserializeCpuTime = if (threadMXBean.isCurrentThreadCpuTimeSupported) {
threadMXBean.getCurrentThreadCpuTime - deserializeStartCpuTime
} else 0L
var writer: ShuffleWriter[Any, Any] = null
try {
val manager = SparkEnv.get.shuffleManager
writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context)
// 执行计算,并将结果写入本地系统的BlockManager中
writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
// 关闭writer,返回计算结果
// 返回包含了数据的location和size元数据信息的MapStatus信息
writer.stop(success = true).get
} catch {
}
}
ShuffleMapTask会将计算结果写入到BlockManager中,最终会返回包含相关元数据信息的MapStatus。MapStatus将成为下一阶段获取输入数据时的依据。
再看看ResultTask的runTask的实现:
override def runTask(context: TaskContext): U = {
// Deserialize the RDD and the func using the broadcast variables.
val threadMXBean = ManagementFactory.getThreadMXBean
val deserializeStartTime = System.currentTimeMillis()
val deserializeStartCpuTime = if (threadMXBean.isCurrentThreadCpuTimeSupported) {
threadMXBean.getCurrentThreadCpuTime
} else 0L
val ser = SparkEnv.get.closureSerializer.newInstance()
// 反序列化
val (rdd, func) = ser.deserialize[(RDD[T], (TaskContext, Iterator[T]) => U)](
ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)
_executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime
_executorDeserializeCpuTime = if (threadMXBean.isCurrentThreadCpuTimeSupported) {
threadMXBean.getCurrentThreadCpuTime - deserializeStartCpuTime
} else 0L
// 执行func进行计算
func(context, rdd.iterator(partition, context))
}
ResultTask会直接执行封装进来的func函数,返回计算结果。
执行完成后,调用CoarseGrainedExecutorBackend.statusUpdate()。statusUpdate方法中向Driver终端点发送StatusUpdate的消息汇报任务执行结果。
结果处理
Driver接到StatusUpdate消息后,调用TaskSchedulerImpl.statusUpdate()进行处理
override def receive: PartialFunction[Any, Unit] = {
case StatusUpdate(executorId, taskId, state, data) =>
// 调用statusUpdate处理
scheduler.statusUpdate(taskId, state, data.value)
if (TaskState.isFinished(state)) {
executorDataMap.get(executorId) match {
case Some(executorInfo) =>
executorInfo.freeCores += scheduler.CPUS_PER_TASK
makeOffers(executorId)
case None =>
》。
}
}
}
看看statusUpdate方法:
def statusUpdate(tid: Long, state: TaskState, serializedData: ByteBuffer) {
var failedExecutor: Option[String] = None
var reason: Option[ExecutorLossReason] = None
synchronized {
try {
taskIdToTaskSetManager.get(tid) match {
case Some(taskSet) =>
// 如果FINISHED,调用taskResultGetter.enqueueSuccessfulTask()
if (TaskState.isFinished(state)) {
cleanupTaskState(tid)
taskSet.removeRunningTask(tid)
if (state == TaskState.FINISHED) {
taskResultGetter.enqueueSuccessfulTask(taskSet, tid, serializedData)
} else if (Set(TaskState.FAILED, TaskState.KILLED, TaskState.LOST).contains(state)) {
taskResultGetter.enqueueFailedTask(taskSet, tid, state, serializedData)
}
}
case None =>
....
}
} catch {
}
}
}
statusUpdate内部会根据任务的状态不同做不同处理,这里只说一下任务是FINISHED的情况。
如果状态是TaskState.FINISHED,调用TaskResultGetter的enqueueSuccessfulTask方法处理。
enqueueSuccessfulTask内部根据结果类型进行处理。如果是IndirectTaskResult,通过blockid从远程获取结果;如果DirectTaskResult,那么无需远程获取。
如果任务是ShuffleMapTask,需要将结果告知下游调度阶段,以便作为后续调度阶段的输入。
这个是在DAGScheduler的handleTaskCompletion中实现的,将MapStatus注册到MapOutputTrackerMaster中,从而完成ShuffleMapTask的处理
如果任务是ResultTask,如果完成,直接标记作业已经完成。
至此整个流程就走了一遍了。
在任务资源分配和结果处理说的有点不清晰,但对于了解整个任务执行流程没有很大影响。
end.
以上是结合看书以及看源码写的流程,如有偏差,欢迎交流指正。
Reference
《图解Spark核心技术与案例实战》
个人公众号:码农峰,定时推送行业资讯,持续发布原创技术文章,欢迎大家关注。