DAGScheduler 源码走读

dagScheduler.runJob(rdd,cleanedFunc,partitions,callSite,resultHandler,localProperties.get)

DAGScheduler中的runJob()会调用submit()：

submitJob(rdd,func,partitions,callSite,resultHandler,properties)

而submitJob()会把job提交到eventProcessLoop线程：

eventProcessLoop.post(JobSubmitted(jobId,rdd,func2,partitions.toArray,callSite,waiter,SerializationUtils.clone(properties)))

类DAGSchedulerEventProcessLoop有一个监听方法onReceive，这个方法会调用doOnReceive方法处理各种case，其代码如下：

defdoOnReceive(event:DAGSchedulerEvent):Unit=eventmatch{

caseJobSubmitted(...)=>

dagScheduler.handleJobSubmitted(...)

......//此处省略了其它无关代码

}

而handleJobSubmitted()方法，首先创建finalStage，最终调用submitStage方法提交创建好的finalStage，其主要代码如下：

defhandleJobSubmitted(...):{

//创建finalStage

finalStage=newResultStage(

//创建job

valjob=newActiveJob(jobId,finalStage,callSite,listener,properties)

//创建好的job加到finalStage中

finalStage.setActiveJob(job)

//最后提交创建的finalStage

		submitStage(finalStage)

......//此处省略其它代码}

接下来看看submitStage方法做了哪些事情：

privatedefsubmitStage(stage:Stage){

valjobId=activeJobForStage(stage)

if(jobId.isDefined){

logDebug("submitStage("+stage+")")

if(!waitingStages(stage)&&!runningStages(stage)&&!failedStages(stage)){

valmissing=getMissingParentStages(stage).sortBy(_.id)//根据传入finalStage查看是否有遗漏的parentstage

logDebug("missing:"+missing)

if(missing.isEmpty){//如果所有parentstage都已经提交，则调用submitMissingTasks方法

logInfo("Submitting"+stage+"("+stage.rdd+"),whichhasnomissingparents")

submitMissingTasks(stage,jobId.get)

}else{

for(parent<-missing){

submitStage(parent)//如果有遗漏的parentstage，则递归调用submitStage方法，使得parentstage先被处理

}

waitingStages+=stage

}

}else{

}

abortStage(stage,"Noactivejobforstage"+stage.id,None)

}

}

privatedefgetMissingParentStages(stage:Stage):List[Stage]={valmissing=newHashSet[Stage]valvisited=newHashSet[RDD[_]]valwaitingForVisit=newStack[RDD[_]]defvisit(rdd:RDD[_]){if(!visited(rdd)){visited+=rddvalrddHasUncachedPartitions=getCacheLocs(rdd).contains(Nil)if(rddHasUncachedPartitions){for(dep<-rdd.dependencies){depmatch{caseshufDep:ShuffleDependency[_,_,_]=>valmapStage=getShuffleMapStage(shufDep,stage.firstJobId)//如果是shuffledependency，则对相应的finalStage生成parentstageif(!mapStage.isAvailable){missing+=mapStage}casenarrowDep:NarrowDependency[_]=>waitingForVisit.push(narrowDep.rdd)//如果是窄依赖的话，则不生成新的stage}}}}}waitingForVisit.push(stage.rdd)while(waitingForVisit.nonEmpty){visit(waitingForVisit.pop())}missing.toList}

privatedefsubmitMissingTasks(stage:Stage,jobId:Int){runningStages+=stagestagematch{cases:ShuffleMapStage=>outputCommitCoordinator.stageStart(stage=s.id,maxPartitionId=s.numPartitions-1)cases:ResultStage=>outputCommitCoordinator.stageStart(stage=s.id,maxPartitionId=s.rdd.partitions.length-1)}valtaskIdToLocations:Map[Int,Seq[TaskLocation]]=try{stagematch{cases:ShuffleMapStage=>partitionsToCompute.map{id=>(id,getPreferredLocs(stage.rdd,id))}.toMapcases:ResultStage=>valjob=s.activeJob.getpartitionsToCompute.map{id=>valp=s.partitions(id)(id,getPreferredLocs(stage.rdd,p))}.toMap}}catch{......}

//TODO:MaybewecankeepthetaskBinaryinStagetoavoidserializingitmultipletimes.//Broadcastedbinaryforthetask,usedtodispatchtaskstoexecutors.Notethatwebroadcast//theserializedcopyoftheRDDandforeachtaskwewilldeserializeit,whichmeanseach//taskgetsadifferentcopyoftheRDD.Thisprovidesstrongerisolationbetweentasksthat//mightmodifystateofobjectsreferencedintheirclosures.ThisisnecessaryinHadoop//wheretheJobConf/Configurationobjectisnotthread-safe.vartaskBinary:Broadcast[Array[Byte]]=nulltry{//ForShuffleMapTask,serializeandbroadcast(rdd,shuffleDep).//ForResultTask,serializeandbroadcast(rdd,func).valtaskBinaryBytes:Array[Byte]=stagematch{casestage:ShuffleMapStage=>closureSerializer.serialize((stage.rdd,stage.shuffleDep):AnyRef).array()casestage:ResultStage=>closureSerializer.serialize((stage.rdd,stage.func):AnyRef).array()}taskBinary=sc.broadcast(taskBinaryBytes)}catch{......}valtasks:Seq[Task[_]]=try{stagematch{casestage:ShuffleMapStage=>partitionsToCompute.map{id=>vallocs=taskIdToLocations(id)valpart=stage.rdd.partitions(id)newShuffleMapTask(stage.id,stage.latestInfo.attemptId,taskBinary,part,locs,stage.internalAccumulators)}casestage:ResultStage=>valjob=stage.activeJob.getpartitionsToCompute.map{id=>valp:Int=stage.partitions(id)valpart=stage.rdd.partitions(p)vallocs=taskIdToLocations(id)newResultTask(stage.id,stage.latestInfo.attemptId,taskBinary,part,locs,id,stage.internalAccumulators)}}}catch{......}if(tasks.size>0){stage.pendingPartitions++=tasks.map(_.partitionId)taskScheduler.submitTasks(newTaskSet(tasks.toArray,stage.id,stage.latestInfo.attemptId,jobId,properties))stage.latestInfo.submissionTime=Some(clock.getTimeMillis())}else{//BecausewepostedSparkListenerStageSubmittedearlier,weshouldmark//thestageascompletedhereincasetherearenotaskstorunmarkStageAsFinished(stage,None)}}