Task的运行过程分析
Task的运行通过Worker启动时生成的Executor实例进行，
caseRegisteredExecutor(sparkProperties)=>
logInfo("Successfullyregistered with driver")
//Make this host instead of hostPort ?
executor=
newExecutor(executorId, Utils.parseHostPort(hostPort)._1,sparkProperties)
通过executor实例的launchTask启动task的运行操作。
deflaunchTask(context: ExecutorBackend,
taskId: Long, serializedTask:ByteBuffer) {
valtr
= newTaskRunner(context, taskId, serializedTask)
runningTasks.put(taskId,tr)
threadPool.execute(tr)
}
生成TaskRunner线程，把task与当前的Wroker的启动的executorBackend传入，
onyarn模式为CoarseGrainedExecutorBackend.
通过threadPool线程池运行生成TaskRunner线程。
TaskRunner.run函数:
用于运行task任务的线程
overridedefrun():
Unit = SparkHadoopUtil.get.runAsUser(sparkUser){ () =>
valstartTime
=System.currentTimeMillis()
SparkEvn后面在进行分析。
SparkEnv.set(env)
Thread.currentThread.setContextClassLoader(replClassLoader)
valser
=SparkEnv.get.closureSerializer.newInstance()
logInfo("Runningtask ID " + taskId)
通过execBackend更新此task的状态。设置task的状态为RUNNING.向master发送StatusUpdate事件。
execBackend.statusUpdate(taskId,TaskState.RUNNING,EMPTY_BYTE_BUFFER)
varattemptedTask:Option[Task[Any]]
= None
vartaskStart:Long
=
defgcTime =ManagementFactory.getGarbageCollectorMXBeans.map(_.getCollectionTime).sum
valstartGCTime=
gcTime
try{
SparkEnv.set(env)
Accumulators.clear()
解析出task的资源信息。包含要运行的jar,其他资源，task定义信息
val(taskFiles,taskJars,taskBytes)=
Task.deserializeWithDependencies(serializedTask)
更新资源信息，并把task运行的jar更新到当前Thread的ClassLoader中。
updateDependencies(taskFiles,taskJars)
通过SparkEnv中配置的Serialize实现对task定义进行反serialize,得到Task实例。
Task的详细实现为ShuffleMapTask或者ResultTask
task=
ser.deserialize[Task[Any]](taskBytes,Thread.currentThread.getContextClassLoader)
假设killed的值为true,不运行当前task任务，进入catch处理。
//If this task has been killed before we
deserializedit, let's quit now. Otherwise,
//continue executing the task.
if(killed)
{
//Throw an exception rather than returning, because returning within atry{} block
//causes a NonLocalReturnControl exception to be thrown. TheNonLocalReturnControl
//exception will be caught by the catch block, leading to an incorrectExceptionFailure
//for the task.
throwTaskKilledException
}
attemptedTask= Some(task)
logDebug("Task" + taskId +"'sepoch
is " + task.epoch)
env.mapOutputTracker.updateEpoch(task.epoch)
生成TaskContext实例，通过Task.runTask运行task的任务，等待task运行完毕。
//Run the actual task and measure its runtime.
taskStart= System.currentTimeMillis()
valvalue
=task.run(taskId.toInt)
valtaskFinish=
System.currentTimeMillis()
此时task运行结束，检查假设task是被killed的结果，进入catch处理。
//If the task has been killed, let's fail it.
if(task.killed){
throwTaskKilledException
}
对task运行的返回结果进行serialize操作。
valresultSer
=SparkEnv.get.serializer.newInstance()
valbeforeSerialization=
System.currentTimeMillis()
valvalueBytes=
resultSer.serialize(value)
valafterSerialization=
System.currentTimeMillis()
发送监控指标
for(m
<-task.metrics){
m.hostname=
Utils.localHostName()
m.executorDeserializeTime=
(taskStart-
startTime).toInt
m.executorRunTime=
(taskFinish-
taskStart).toInt
m.jvmGCTime=
gcTime - startGCTime
m.resultSerializationTime=
(afterSerialization-
beforeSerialization).toInt
}
valaccumUpdates=
Accumulators.values
把Task的返回结果生成DirectTaskResult实例。并对其进行serialize操作。
valdirectResult=
newDirectTaskResult(valueBytes,accumUpdates,task.metrics.getOrElse(null))
valserializedDirectResult=
ser.serialize(directResult)
logInfo("Serializedsize of result for " + taskId +
"is " +serializedDirectResult.limit)
检查taskresult的大小是否超过了akka的发送消息大小，
假设是通过BlockManager来管理结果，设置RDD的存储级别为MEMORY与DISK，否则表示没有达到actor消息大小，
直接使用TaskResult,此部分信息主要是须要通过状态更新向Scheduler向送StatusUpdate事件调用。
valserializedResult=
{
if(serializedDirectResult.limit>=
akkaFrameSize-
) {
logInfo("Storingresult for " + taskId +
"in local BlockManager")
valblockId
=TaskResultBlockId(taskId)
env.blockManager.putBytes(
blockId,serializedDirectResult,StorageLevel.MEMORY_AND_DISK_SER)
ser.serialize(newIndirectTaskResult[Any](blockId))
}
else{
logInfo("Sendingresult for " + taskId +
"directly to driver")
serializedDirectResult
}
}
通过execBackend更新此task的状态。设置task的状态为FINISHED.向master发送StatusUpdate事件。
execBackend.statusUpdate(taskId,TaskState.FINISHED,serializedResult)
logInfo("Finishedtask ID " + taskId)
}
catch{
出现异常，发送FAILED事件。
caseffe:FetchFailedException
=> {
valreason
=ffe.toTaskEndReason
execBackend.statusUpdate(taskId,TaskState.FAILED,ser.serialize(reason))
}
caseTaskKilledException => {
logInfo("Executorkilled task " + taskId)
execBackend.statusUpdate(taskId,TaskState.KILLED,ser.serialize(TaskKilled))
}
caset:Throwable
=> {
valserviceTime=
(System.currentTimeMillis() - taskStart).toInt
valmetrics
=attemptedTask.flatMap(t=> t.metrics)
for(m
<-metrics) {
m.executorRunTime=
serviceTime
m.jvmGCTime=
gcTime - startGCTime
}
valreason
=ExceptionFailure(t.getClass.getName,t.toString,t.getStackTrace,metrics)
execBackend.statusUpdate(taskId,TaskState.FAILED,ser.serialize(reason))
//TODO: Should we exit the whole executor here? On the one hand, thefailed task may
//have left some weird state around depending on when the exception wasthrown, but on
//the other hand, maybe we could detect that when future tasks fail andexit then.
logError("Exceptionin task ID " + taskId,
t)
//System.exit(1)
}
}
finally{
从shuffleMemoryMap中移出此线程相应的shuffle的内存空间
//TODO: Unregister shuffle memory only for ResultTask
valshuffleMemoryMap=
env.shuffleMemoryMap
shuffleMemoryMap.synchronized{
shuffleMemoryMap.remove(Thread.currentThread().getId)
}
从runningTasks中移出此task
runningTasks.remove(taskId)
}
}
}
Task运行过程的状态更新
ExecutorBackend.statusUpdate
onyarn模式实现类CoarseGrainedExecutorBackend,通过master的actor发送StatusUpdate事件。
overridedefstatusUpdate(taskId:
Long, state: TaskState, data: ByteBuffer) {
driver! StatusUpdate(executorId, taskId, state,
data)
}
master中的ExecutorBackend处理状态更新操作：
实现类：CoarseGrainedSchedulerBackend.DriverActor
caseStatusUpdate(executorId,taskId,state,data)
=>
通过TaskSchedulerImpl的statusUpdate处理状态更新。
scheduler.statusUpdate(taskId,state,data.value)
假设Task状态为完毕状态，完毕状态包括(FINISHED,FAILED,KILLED,LOST)
if(TaskState.isFinished(state)){
if(executorActor.contains(executorId)){
每个task占用一个cpucore,此时task完毕，把可用的core值加一
freeCores(executorId) +=
在此executor上接着运行其于的task任务，此部分可參见scheduler调度过程分析中的部分说明。
makeOffers(executorId)
}
else{
//Ignoring the update since we don't know about the executor.
valmsg
="Ignored task status update (%dstate %s) from unknown executor %s with ID %s"
logWarning(msg.format(taskId,state,sender,
executorId))
}
}
TaskSchedulerImpl.statusUpdate函数处理流程
defstatusUpdate(tid: Long, state: TaskState,
serializedData: ByteBuffer){
varfailedExecutor:Option[String]
= None
synchronized {
try{
假设Task的状态传入为Task的运行丢失，同一时候task在executor列表中存在
if(state == TaskState.LOST&&
taskIdToExecutorId.contains(tid)){
得到此task运行的worker所属的executorID，
//We lost this entire executor, so remember that it's gone
valexecId
=taskIdToExecutorId(tid)
假设此executor是active的Executor，运行scheduler的executorLost操作。
包括TaskSetManager，会运行TaskSetManager.executorLost操作.
设置当前的executor为failedExecutor,共函数最后使用。
if(activeExecutorIds.contains(execId)){
removeExecutor(execId)
failedExecutor= Some(execId)
}
}
taskIdToTaskSetId.get(tid)match{
caseSome(taskSetId)=>
假设task状态是完毕状态，非RUNNING状态。移出相应的容器中的值
if(TaskState.isFinished(state)) {
taskIdToTaskSetId.remove(tid)
if(taskSetTaskIds.contains(taskSetId)){
taskSetTaskIds(taskSetId)-=
tid
}
taskIdToExecutorId.remove(tid)
}
activeTaskSets.get(taskSetId).foreach{
taskSet =>
假设task是成功完毕，从TaskSet中移出此task，同一时候通过TaskResultGetter获取数据。
if(state == TaskState.FINISHED){
taskSet.removeRunningTask(tid)
taskResultGetter.enqueueSuccessfulTask(taskSet,tid,
serializedData)
}
elseif(Set(TaskState.FAILED,TaskState.KILLED,TaskState.LOST).contains(state)){
task任务运行失败的处理部分：
taskSet.removeRunningTask(tid)
taskResultGetter.enqueueFailedTask(taskSet,tid, state,
serializedData)
}
}
caseNone =>
logInfo("Ignoringupdate with state %s from TID %s because its task set is gone"
.format(state, tid))
}
}
catch{
casee:Exception
=> logError("Exceptionin statusUpdate",
e)
}
}
假设有failed的worker,通过dagScheduler处理此executor.
//Update the DAGScheduler without holding a lock on this, since thatcan deadlock
if(failedExecutor!=
None) {
dagScheduler.executorLost(failedExecutor.get)
发起task运行的分配与任务运行操作。
backend.reviveOffers()
}
}
TaskStatus.LOST状态,同一时候executor在activeExecutorIds中
TaskStatus的状态为LOST时，同一时候executor是活动的executor(也就是有过运行task的情况)
privatedefremoveExecutor(executorId:
String) {
从activeExecutorIds中移出此executor
activeExecutorIds-= executorId
得到此executor相应的worker的host
valhost
=executorIdToHost(executorId)
取出host相应的全部executor,并移出当前的executor
valexecs
=executorsByHost.getOrElse(host,newHashSet)
execs-=
executorId
if(execs.isEmpty){
executorsByHost-=
host
}
从executor相应的host容器中移出此executor
executorIdToHost-= executorId
此处主要是去运行TaskSetManager.executorLost函数。
rootPool.executorLost(executorId,host)
}
TaskSetManager.executorLost函数:
此函数主要处理executor导致task丢失的情况，把executor上的task又一次加入到pending的tasks列表中
overridedefexecutorLost(execId:
String, host: String) {
logInfo("Re-queueingtasks for " + execId +
"from TaskSet " +
taskSet.id)
//Re-enqueue pending tasks for this host based on the status of thecluster -- for example,
a
//task that used to have locations on only this host might now go tothe no-prefslist.
Note
//that it's okay if we add a task to the same queue twice (if it hadmultiple preferred
//locations), because findTaskFromList will skip already-running tasks.
又一次生成此TaskSet中的pending队列，由于当前executor的实例被移出，须要又一次生成。
for(index
<-getPendingTasksForExecutor(execId)) {
addPendingTask(index,readding=true)
}
for(index
<-getPendingTasksForHost(host)) {
addPendingTask(index,readding=true)
}
//Re-enqueue any tasks that ran on the failed executor if this is ashuffle map stage
假设当前的RDD是shuffle的rdd,
).isInstanceOf[ShuffleMapTask]){
for((tid,info)
<-taskInfosifinfo.executorId==
execId) {
valindex
=taskInfos(tid).index
if(successful(index)){
successful(index)=
false
copiesRunning(index) -=
tasksSuccessful-=
addPendingTask(index)
//Tell the DAGScheduler that this task was resubmitted so that itdoesn't think our
//stage finishes when a total of tasks.size tasks finish.
通过DAGScheduler发送CompletionEvent处理事件，事件类型为Resubmitted,
sched.dagScheduler.taskEnded(tasks(index),Resubmitted,
null,null,info,
null)
}
}
}
假设task还处于running状态，同一时候此task在lost的executor上执行，
//Also re-enqueue any tasks that were running on the node
for((tid,info)
<-taskInfosifinfo.running&&
info.executorId== execId) {
设置task的failed值为true,移出此task的running列表中的值，又一次加入task到pendingtasks队列中。
handleFailedTask(tid,TaskState.FAILED,None)
}
}
DAGScheduler处理CompletionEvent事件。
...........................
casecompletion@
CompletionEvent(task,reason,_,
_, taskInfo,taskMetrics)=>
listenerBus.post(SparkListenerTaskEnd(task,reason,taskInfo,taskMetrics))
handleTaskCompletion(completion)
.........................
caseResubmitted =>
logInfo("Resubmitted" +
task+
", so marking it as stillrunning")
pendingTasks(stage)+=
task
(TaskState.FAILED,TaskState.KILLED,TaskState.LOST)状态
.........................
}
elseif(Set(TaskState.FAILED,TaskState.KILLED,TaskState.LOST).contains(state)){
把task从running容器中移出
taskSet.removeRunningTask(tid)
此函数主要是解析出出错的信息。并通过TaskSchedulerImpl.handleFailedTask处理exception
taskResultGetter.enqueueFailedTask(taskSet,tid, state,
serializedData)
}
TaskSchedulerImpl.handleFailedTask函数：
defhandleFailedTask(
taskSetManager: TaskSetManager,
tid: Long,
taskState: TaskState,
reason: Option[TaskEndReason]) =synchronized {
taskSetManager.handleFailedTask(tid,taskState, reason)
假设task不是被KILLED掉的task,又一次发起task的分配与运行操作。
if(taskState != TaskState.KILLED){
//Need to revive offers again now that the task set manager state hasbeen updated to
//reflect failed tasks that need to be re-run.
backend.reviveOffers()
}
}
TaskSetManager.handleFailedTask函数流程
TaskSetManager.handleFailedTask,函数，处理task运行的exception信息。
defhandleFailedTask(tid: Long, state:
TaskState, reason:Option[TaskEndReason]) {
valinfo
=taskInfos(tid)
if(info.failed){
return
}
removeRunningTask(tid)
valindex
=info.index
info.markFailed()
varfailureReason=
"unknown"
if(!successful(index)){
logWarning("LostTID %s (task %s:%d)".format(tid,taskSet.id,index))
copiesRunning(index) -=
假设是通过TaskSetManager.executorLost函数发起的此函数调用(Task.LOST)，以下的case部分不会运行，
否则是task的运行exception情况，也就是状态更新中非Task.LOST状态时。
//Check if the problem is a map output fetch failure. In that case,this
//task will never succeed on any node, so tell the scheduler about it.
reason.foreach {
casefetchFailed:FetchFailed
=>
读取失败，移出全部此taskset的task运行。并从scheduler中移出此taskset的调度,不再运行以下流程
logWarning("Losswas due to fetch failure from " +fetchFailed.bmAddress)
sched.dagScheduler.taskEnded(tasks(index),fetchFailed,null,null,info,
null)
successful(index)=
true
tasksSuccessful+=
sched.taskSetFinished(this)
removeAllRunningTasks()
return
caseTaskKilled =>
task被kill掉，移出此task,同一时候不再运行以下流程
logWarning("Task%d was killed.".format(tid))
sched.dagScheduler.taskEnded(tasks(index),reason.get,
null,null,info,
null)
return
caseef:ExceptionFailure
=>
sched.dagScheduler.taskEnded(
tasks(index),ef,
null,null,info,ef.metrics.getOrElse(null))
if(ef.className==
classOf[NotSerializableException].getName()) {
//If the task result wasn't
rerializable,there's no point in trying to re-execute it.
logError("Task%s:%s had a not serializable result: %s; not retrying".format(
taskSet.id,index,ef.description))
abort("Task%s:%s had a not serializable result: %s".format(
taskSet.id,index,ef.description))
return
}
valkey
=ef.description
failureReason=
"Exception failure:%s".format(ef.description)
valnow
=clock.getTime()
val(printFull,dupCount)
={
if(recentExceptions.contains(key)){
val(dupCount,printTime)=
recentExceptions(key)
if(now
-printTime >EXCEPTION_PRINT_INTERVAL){
recentExceptions(key)=
(,
now)
)
}
else{
recentExceptions(key)=
(dupCount+
,printTime)
(false,dupCount
+)
}
}
else{
recentExceptions(key)=
(,
now)
)
}
}
if(printFull){
vallocs
=ef.stackTrace.map(loc=>
"\tat%s".format(loc.toString))
logWarning("Losswas due to %s\n%s\n%s".format(
ef.className,ef.description,locs.mkString("\n")))
}
else{
logInfo("Losswas due to %s [duplicate %d]".format(ef.description,dupCount))
}
caseTaskResultLost =>
failureReason=
"Lost result for TID %s onhost %s".format(tid,
info.host)
logWarning(failureReason)
sched.dagScheduler.taskEnded(tasks(index),TaskResultLost,
null,null,info,
null)
case_ => {}
}
又一次把task加入到pending的运行队列中,同一时候假设状态非KILLED的状态，设置并检查是否达到重试的最大次数
//On non-fetch failures, re-enqueue the task as pending for a maxnumber of retries
addPendingTask(index)
if(state != TaskState.KILLED){
numFailures(index) +=
if(numFailures(index)>=
maxTaskFailures){
logError("Task%s:%d failed %d times; aborting job".format(
taskSet.id,index,maxTaskFailures))
abort("Task%s:%d failed %d times (most recent failure: %s)".format(
taskSet.id,index,maxTaskFailures,failureReason))
}
}
}else{
logInfo("Ignoringtask-lost event for TID " + tid +
"because task " +
index+
" is already finished")
}
}
DAGScheduler处理taskEnded流程：
deftaskEnded(
task: Task[_],
reason: TaskEndReason,
result: Any,
accumUpdates: Map[Long, Any],
taskInfo: TaskInfo,
taskMetrics: TaskMetrics) {
eventProcessActor! CompletionEvent(task, reason,
result, accumUpdates, taskInfo,taskMetrics)
}
处理CompletionEvent事件：
casecompletion@
CompletionEvent(task,reason, _,_,
taskInfo,taskMetrics)=>
listenerBus.post(SparkListenerTaskEnd(task,reason,taskInfo,taskMetrics))
handleTaskCompletion(completion)
DAGScheduler.handleTaskCompletion
读取失败的case,
caseFetchFailed(bmAddress,shuffleId,mapId,reduceId)=>
//Mark the stage that the reducer was in as
unrunnable
valfailedStage=
stageIdToStage(task.stageId)
running-=
failedStage
failed+=
failedStage
..............................
//Mark the map whose fetch failed as broken in the map stage
valmapStage
=shuffleToMapStage(shuffleId)
if(mapId
!=-) {
mapStage.removeOutputLoc(mapId,bmAddress)
mapOutputTracker.unregisterMapOutput(shuffleId,mapId,bmAddress)
}
...........................
failed+=
mapStage
//Remember that a fetch failed now; this is used to resubmit the broken
//stages later, after a small wait (to give other tasks the chance tofail)
lastFetchFailureTime= System.currentTimeMillis()
// TODO:Use pluggableclock
//TODO: mark the executor as failed only if there were lots of fetchfailures on it
if(bmAddress!=
null){
把stage中可运行的partition中相应的executorid的location所有移出。
handleExecutorLost(bmAddress.executorId,Some(task.epoch))
}
caseExceptionFailure(className,description,stackTrace,metrics)
=>
//Do nothing here, left up to the TaskScheduler to decide how to handleuser failures
caseTaskResultLost =>
//Do nothing here; the TaskScheduler handles these failures andresubmits the task.
TaskStatus.FINISHED状态
此状态表示task正常完毕，
if(state == TaskState.FINISHED){
移出taskSet中的running队列中移出此task
taskSet.removeRunningTask(tid)
获取task的响应数据。
taskResultGetter.enqueueSuccessfulTask(taskSet,tid,
serializedData)
TaskResultGetter.enqueueSuccessfulTask函数：
defenqueueSuccessfulTask(
taskSetManager: TaskSetManager,tid: Long, serializedData: ByteBuffer) {
getTaskResultExecutor.execute(newRunnable
{
overridedefrun()
{
try{
从响应的结果中得到数据，须要先运行deserialize操作。
valresult
=serializer.get().deserialize[TaskResult[_]](serializedData)match{
假设result的结果小于akka的actor传输的大小，直接返回task的运行结果
casedirectResult:DirectTaskResult[_]
=> directResult
caseIndirectTaskResult(blockId)=>
否则，result结果太大，通过BlockManager管理，通过blockManager拿到result的数据
logDebug("Fetchingindirect task result for TID %s".format(tid))
给DAGScheduler发送GettingResultEvent事件处理，
见以下TaskSchedulerImpl.handleTaskGettingResult函数
scheduler.handleTaskGettingResult(taskSetManager,tid)
得到task的运行结果
valserializedTaskResult=
sparkEnv.blockManager.getRemoteBytes(blockId)
task运行完毕，并拿结果失败，见上面的错误处理中的TaskResultLost部分。
if(!serializedTaskResult.isDefined){
/*We won't be able to get the task result if the machine that ran thetask failed
* between when the taskended and when we tried to fetch the result, or if the
* block manager had toflush the result. */
scheduler.handleFailedTask(
taskSetManager, tid,TaskState.FINISHED,Some(TaskResultLost))
return
}
对task的运行结果进行deserialized操作。
valdeserializedResult=
serializer.get().deserialize[DirectTaskResult[_]](
serializedTaskResult.get)
拿到运行结果，移出相应的blockid
sparkEnv.blockManager.master.removeBlock(blockId)
deserializedResult
}
result.metrics.resultSize=
serializedData.limit()
见以下的TaskSchedulerImpl.handleSuccessfulTask处理函数。
scheduler.handleSuccessfulTask(taskSetManager,tid,
result)
}
catch{
casecnf:ClassNotFoundException
=>
valloader
=Thread.currentThread.getContextClassLoader
taskSetManager.abort("ClassNotFoundwith classloader: "
+ loader)
caseex:Throwable
=>
taskSetManager.abort("Exceptionwhile deserializing and fetching task: %s".format(ex))
}
}
})
}
TaskSchedulerImpl.handleTaskGettingResult函数：
defhandleTaskGettingResult(taskSetManager:
TaskSetManager, tid: Long) {
taskSetManager.handleTaskGettingResult(tid)
}
taskSetManager中
defhandleTaskGettingResult(tid: Long)
= {
valinfo
=taskInfos(tid)
info.markGettingResult()
sched.dagScheduler.taskGettingResult(tasks(info.index),info)
}
通过DAGScheduler发起GettingResultEvent事件。
deftaskGettingResult(task: Task[_],
taskInfo: TaskInfo) {
eventProcessActor! GettingResultEvent(task, taskInfo)
}
对GettingResultEvent事件的处理：事实上就是打个酱油，无实际处理操作。
caseGettingResultEvent(task,taskInfo)=>
listenerBus.post(SparkListenerTaskGettingResult(task,taskInfo))
TaskSchedulerImpl.handleSuccessfulTask处理函数：
defhandleSuccessfulTask(
taskSetManager: TaskSetManager,
tid: Long,
taskResult: DirectTaskResult[_])= synchronized {
taskSetManager.handleSuccessfulTask(tid,taskResult)
}
TastSetManager中
defhandleSuccessfulTask(tid: Long,
result: DirectTaskResult[_]) = {
valinfo
=taskInfos(tid)
valindex
=info.index
info.markSuccessful()
从running队列中移出此task
removeRunningTask(tid)
if(!successful(index)){
logInfo("FinishedTID %s in %d ms on %s (progress: %d/%d)".format(
tid,
info.duration,info.host,tasksSuccessful,numTasks))
向dagscheduler发送success消息，
sched.dagScheduler.taskEnded(
tasks(index),Success,
result.value,result.accumUpdates,info,result.metrics)
设置成功完毕的task个数加一,同一时候在successful容器中设置task相应的执行状态为true,表示成功。
//Mark successful and stop if all the tasks have succeeded.
tasksSuccessful+=
successful(index)=
true
假设完毕的task个数，达到task的总个数，完毕此taskset,也就相当于完毕了一个rdd
if(tasksSuccessful==
numTasks){
sched.taskSetFinished(this)
}
}else{
logInfo("Ignorningtask-finished event for TID " + tid+
" because task "+
index+
" has already completedsuccessfully")
}
}
DAGScheduler处理CompletionEvent的Success,,,,
caseSuccess
=>
logInfo("Completed" +
task)
if(event.accumUpdates!=
null){
Accumulators.add(event.accumUpdates)//
TODO: do this only if task wasn'tresubmitted
}
把等待运行队列中移出此task
pendingTasks(stage)-=
task
stageToInfos(stage).taskInfos+=
event.taskInfo-> event.taskMetrics
依据task的运行类型，处理两个类型的Task
taskmatch{
假设task是ResultTask,表示不须要shuffle操作
casert:ResultTask[_,
_] =>
resultStageToJob.get(stage)match{
caseSome(job)=>
假设此运行的stage的ActiveJob中相应此task的partition存储的finished标志为false,
if(!job.finished(rt.outputId)){
设置task的完毕标志为true
job.finished(rt.outputId)=
true
把job中完毕的task个数加一，同一时候检查是否全部的task都完毕,假设全部task都完毕，
从相关的容器中移出此job与相应的stage.
job.numFinished+=
//If the whole job has finished, remove it
if(job.numFinished==
job.numPartitions){
idToActiveJob-=
stage.jobId
activeJobs-=
job
resultStageToJob-=
stage
markStageAsFinished(stage)
jobIdToStageIdsRemove(job.jobId)
listenerBus.post(SparkListenerJobEnd(job,JobSucceeded))
}
调用ActiveJob内的JobWaiter.taskSucceeded函数，更新此task为完毕，同一时候把result传入进行输出处理。
job.listener.taskSucceeded(rt.outputId,event.result)
}
caseNone =>
logInfo("Ignoringresult from " +
rt+
" because its job has finished")
}
针对shuffle的task的运行完毕，处理流程：
casesmt:ShuffleMapTask
=>
valstatus
=event.result.asInstanceOf[MapStatus]
valexecId
=status.location.executorId
logDebug("ShuffleMapTaskfinished on " +
execId)
if(failedEpoch.contains(execId)&&
smt.epoch<=
failedEpoch(execId)){
logInfo("Ignoringpossibly bogus ShuffleMapTask completion from "+
execId)
}
else{
把shuffle的result(MapStatus)写入到stage的outputLoc中。每加入一个会把numAvailableOutputs的值加一，
当numAvailableOutputs的值==numPartitions的值时，表示shuffle的map运行完毕。
stage.addOutputLoc(smt.partitionId,status)
}
假设此stage还处在running状态，同一时候pendingTasks中全部的task已经处理完毕
if(running.contains(stage)&&
pendingTasks(stage).isEmpty){
更新stage的状态
markStageAsFinished(stage)
.......................................
此处表示shuffle的stage处理完毕，把shuffleid与stage的outputLocs注冊到mapOutputTracker中。
把每个shuffletaks运行的executor与host等信息，每个task运行完毕的大小。注冊到mapoutput中。
每个task的shuffle的writer都会有shuffleid的信息，注冊成功后，
下一个stage会依据mapoutputtracker中此shuffleid的信息读取数据。
mapOutputTracker.registerMapOutputs(
stage.shuffleDep.get.shuffleId,
stage.outputLocs.map(list=>
if(list.isEmpty)
nullelselist.head).toArray,
changeEpoch =
true)
}
clearCacheLocs()
stage中每个partition的outputLoc默认值为Nil，假设发现有partition的值为Nil,表示有task处理失败，
又一次提交此stage.此时会把没有成功的task又一次运行。
if(stage.outputLocs.exists(_==
Nil)) {
.........................................
submitStage(stage)
}
else{
valnewlyRunnable=
newArrayBuffer[Stage]
for(stage
<-waiting) {
logInfo("Missingparents for " +
stage+
": "+ getMissingParentStages(stage))
}
此处检查以下未运行的全部的stage,假设stage(RDD)的上级shuffle依赖完毕,
或者后面全部的stage不再有shuffle的stage的全部stage,拿到这些个stage.
for(stage
<-waiting
ifgetMissingParentStages(stage) == Nil) {
newlyRunnable+=
stage
}
运行此stage后面的全部可运行的stage,把waiting中移出要运行的stage,
waiting--=
newlyRunnable
在running队列中加入要运行的新的stage.
running++=
newlyRunnable
for{
stage<-
newlyRunnable.sortBy(_.id)
jobId<- activeJobForStage(stage)
} {
提交下一个stage的task分配与运行。
logInfo("Submitting" +
stage+
" ("+ stage.rdd+
"), which is now runnable")
submitMissingTasks(stage,jobId)
}
}
}
}
JobWaiter.taskSucceeded函数，
task完毕后的处理函数。
overridedef taskSucceeded(index:
Int,result: Any): Unit = synchronized {
if(_jobFinished){
thrownewUnsupportedOperationException("taskSucceeded()called
on a finished JobWaiter")
}
通过resultHandler函数把结果进行处理。此函数是生成JobWaiter时传入
resultHandler(index,result.asInstanceOf[T])
把完毕的task值加一
finishedTasks+=
if(finishedTasks==
totalTasks) {
假设完毕的task个数等于全部的task的个数时，设置job的完毕状态为true,并设置状态为JobSucceeded
假设设置为true,表示job运行完毕，前面的等待运行完毕结束等待。
_jobFinished=
true
jobResult= JobSucceeded
this.notifyAll()
}
}
Task.runTask函数实现
Task的实现分为两类，
须要进行shuffle操作的ShuffleMapTask,
不须要进行shuffle操作的ResultTask.
ResulitTask.runTask
overridedef runTask(context:TaskContext):
U = {
metrics= Some(context.taskMetrics)
try{
此处通过生成task实例时也就是DAGScheduler的runJob时传入的function进行处理
比方在PairRDDFunction.saveAsHadoopDataset中定义的writeToFile函数
rdd.iterator中会依据不现的RDD的实现，运行其compute函数，
而compute函数详细运行通过业务代码中定义的如map函数传入的定义的function进行运行，
func(context,rdd.iterator(split,context))
}finally{
context.executeOnCompleteCallbacks()
}
}
ShuffleMapTask.runTask
overridedef runTask(context:TaskContext):
MapStatus = {
valnumOutputSplits=
dep.partitioner.numPartitions
metrics= Some(context.taskMetrics)
valblockManager=
SparkEnv.get.blockManager
valshuffleBlockManager=
blockManager.shuffleBlockManager
varshuffle:ShuffleWriterGroup
= null
varsuccess
=false
try{
通过shuffleId拿到一个shuffle的写入实例
//Obtain all the block writers for shuffle blocks.
valser
=SparkEnv.get.serializerManager.get(dep.serializerClass,SparkEnv.get.conf)
shuffle=
shuffleBlockManager.forMapTask(dep.shuffleId,partitionId,numOutputSplits,ser)
运行rdd.iterator操作，生成Pair,也就是Product2,依据key又一次shuffle到不同的文件里。
当全部的shuffle的task完毕后，会把此stage注冊到mapOutputTracker中，
等待下一个stage从中读取数据并运行其他操作，每个shuffle的task完毕后会生成一个MapStatus实例，
此实例主要包括有shuffle运行的executor与host等信息，每个task运行完毕的大小。
详细的shuffle数据读取可參见后面的shufle分析.
//Write the map output to its associated buckets.
for(elem
<-rdd.iterator(split,context)) {
valpair
=elem.asInstanceOf[Product2[Any,Any]]
valbucketId
=dep.partitioner.getPartition(pair._1)
shuffle.writers(bucketId).write(pair)
}
//Commit the writes. Get the size of each bucket block (total blocksize).
vartotalBytes=
0L
vartotalTime
=0L
valcompressedSizes:Array[Byte]
= shuffle.writers.map{ writer: BlockObjectWriter =>
writer.commit()
writer.close()
valsize
=writer.fileSegment().length
totalBytes+=
size
totalTime+= writer.timeWriting()
MapOutputTracker.compressSize(size)
}
//Update shuffle metrics.
valshuffleMetrics=
newShuffleWriteMetrics
shuffleMetrics.shuffleBytesWritten=
totalBytes
shuffleMetrics.shuffleWriteTime=
totalTime
metrics.get.shuffleWriteMetrics=
Some(shuffleMetrics)
success=
true
newMapStatus(blockManager.blockManagerId,compressedSizes)
}catch{
casee:Exception =>
//If there is an exception from running the task, revert the partialwrites
//and throw the exception upstream to Spark.
if(shuffle
!=null&&
shuffle.writers!=
null){
for(writer
<-shuffle.writers){
writer.revertPartialWrites()
writer.close()
}
}
throwe
}finally{
//Release the writers back to the shuffle block manager.
if(shuffle
!=null&&
shuffle.writers!=
null){
shuffle.releaseWriters(success)
}
//Execute the
callbackson task completion.
context.executeOnCompleteCallbacks()
}
}