Spark源码分析之-scheduler模块

Background

Spark在资源管理和调度方式上采用了类似于Hadoop YARN的方式，最上层是资源调度器，它负责分配资源和调度注册到Spark中的所有应用，Spark选用Mesos或是YARN等作为其资源调度框架。在每一个应用内部，Spark又实现了任务调度器，负责任务的调度和协调，类似于MapReduce。本质上，外层的资源调度和内层的任务调度相互独立，各司其职。本文对于Spark的源码分析主要集中在内层的任务调度器上，分析Spark任务调度器的实现。

Scheduler模块整体架构

scheduler模块主要分为两大部分：

1. TaskSchedulerListener。TaskSchedulerListener部分的主要功能是监听用户提交的job，将job分解为不同的类型的stage以及相应的task，并向TaskScheduler提交task。
2. TaskScheduler。TaskScheduler接收用户提交的task并执行。而TaskScheduler根据部署的不同又分为三个子模块:
   • ClusterScheduler
   • LocalScheduler
   • MesosScheduler

TaskSchedulerListener

Spark抽象了TaskSchedulerListener并在其上实现了DAGScheduler。DAGScheduler的主要功能是接收用户提交的job，将job根据类型划分为不同的stage，并在每一个stage内产生一系列的task，向TaskScheduler提交task。下面我们首先来看一下TaskSchedulerListener部分的类图：

• 用户所提交的job在得到DAGScheduler的调度后，会被包装成ActiveJob，同时会启动JobWaiter阻塞监听job的完成状况。
• 于此同时依据job中RDD的dependency和dependency属性(NarrowDependency，ShufflerDependecy)，DAGScheduler会根据依赖关系的先后产生出不同的stage DAG(result stage, shuffle map stage)。
• 在每一个stage内部，根据stage产生出相应的task，包括ResultTask或是ShuffleMapTask，这些task会根据RDD中partition的数量和分布，产生出一组相应的task，并将其包装为TaskSet提交到TaskScheduler上去。

RDD的依赖关系和Stage的分类

在Spark中，每一个RDD是对于数据集在某一状态下的表现形式，而这个状态有可能是从前一状态转换而来的，因此换句话说这一个RDD有可能与之前的RDD(s)有依赖关系。根据依赖关系的不同，可以将RDD分成两种不同的类型：Narrow Dependency和Wide Dependency。

• Narrow Dependency指的是 child RDD只依赖于parent RDD(s)固定数量的partition。
• Wide Dependency指的是child RDD的每一个partition都依赖于parent RDD(s)所有partition。

它们之间的区别可参看下图：

根据RDD依赖关系的不同，Spark也将每一个job分为不同的stage，而stage之间的依赖关系则形成了DAG。对于Narrow Dependency，Spark会尽量多地将RDD转换放在同一个stage中；而对于Wide Dependency，由于Wide Dependency通常意味着shuffle操作，因此Spark会将此stage定义为ShuffleMapStage，以便于向MapOutputTracker注册shuffle操作。对于stage的划分可参看下图，Spark通常将shuffle操作定义为stage的边界。

DAGScheduler

在用户创建SparkContext对象时，Spark会在内部创建DAGScheduler对象，并根据用户的部署情况，绑定不同的TaskSechduler，并启动DAGcheduler

1. privatevar taskScheduler:TaskScheduler={
2. //...
3. }
4. taskScheduler.start()
6. privatevar dagScheduler =newDAGScheduler(taskScheduler)
7. dagScheduler.start()

而DAGScheduler的启动会在内部创建daemon线程，daemon线程调用run()从block queue中取出event进行处理。

1. privatedef run(){
2. SparkEnv.set(env)
4. while(true){
5. val event= eventQueue.poll(POLL_TIMEOUT,TimeUnit.MILLISECONDS)
6. if(event!=null){
7. logDebug("Got event of type "+event.getClass.getName)
8. }
10. if(event!=null){
11. if(processEvent(event)){
12. return
13. }
14. }
16. val time =System.currentTimeMillis()// TODO: use a pluggable clock for testability
17. if(failed.size >0&& time > lastFetchFailureTime + RESUBMIT_TIMEOUT){
18. resubmitFailedStages()
19. }else{
20. submitWaitingStages()
21. }
22. }
23. }

而run()会调用processEvent来处理不同的event。

DAGScheduler处理的event包括：

• JobSubmitted
• CompletionEvent
• ExecutorLost
• TaskFailed
• StopDAGScheduler

根据event的不同调用不同的方法去处理。

本质上DAGScheduler是一个生产者-消费者模型，用户和TaskSchduler产生event将其放入block queue，daemon线程消费event并处理相应事件。

Job的生与死

既然用户提交的job最终会交由DAGScheduler去处理，那么我们就来研究一下DAGScheduler处理job的整个流程。在这里我们分析两种不同类型的job的处理流程。

1. 没有shuffle和reduce的job

   1. val textFile = sc.textFile("README.md")
   2. textFile.filter(line => line.contains("Spark")).count()

2. 有shuffle和reduce的job

   1. val textFile = sc.textFile("README.md")
   2. textFile.flatMap(line => line.split(" ")).map(word =>(word,1)).reduceByKey((a, b)=> a + b)

首先在对RDD的count()和reduceByKey()操作都会调用SparkContext的runJob()来提交job，而SparkContext的runJob()最终会调用DAGScheduler的runJob()：

1. def runJob[T, U:ClassManifest](
2. finalRdd: RDD[T],
3. func:(TaskContext,Iterator[T])=> U,
4. partitions:Seq[Int],
5. callSite:String,
6. allowLocal:Boolean,
7. resultHandler:(Int, U)=>Unit)
8. {
9. if(partitions.size ==0){
10. return
11. }
12. val (toSubmit, waiter)= prepareJob(
13. finalRdd, func, partitions, callSite, allowLocal, resultHandler)
14. eventQueue.put(toSubmit)
15. waiter.awaitResult() match {
16. caseJobSucceeded=>{}
17. caseJobFailed(exception:Exception)=>
18. logInfo("Failed to run "+ callSite)
19. throw exception
20. }
21. }

runJob()会调用prepareJob()对job进行预处理，封装成JobSubmitted事件，放入queue中，并阻塞等待job完成。

当daemon线程的processEvent()从queue中取出JobSubmitted事件后，会根据job划分出不同的stage，并且提交stage：

1. caseJobSubmitted(finalRDD, func, partitions, allowLocal, callSite, listener)=>
2. val runId = nextRunId.getAndIncrement()
3. val finalStage = newStage(finalRDD,None, runId)
4. val job =newActiveJob(runId, finalStage, func, partitions, callSite, listener)
5. clearCacheLocs()
6. if(allowLocal && finalStage.parents.size ==0&& partitions.length ==1){
7. runLocally(job)
8. }else{
9. activeJobs += job
10. resultStageToJob(finalStage)= job
11. submitStage(finalStage)
12. }

首先，对于任何的job都会产生出一个finalStage来产生和提交task。其次对于某些简单的job，它没有依赖关系，并且只有一个partition，这样的job会使用local thread处理而并非提交到TaskScheduler上处理。

接下来产生finalStage后，需要调用submitStage()，它根据stage之间的依赖关系得出stage DAG，并以依赖关系进行处理：

1. privatedef submitStage(stage:Stage){
2. if(!waiting(stage)&&!running(stage)&&!failed(stage)){
3. val missing = getMissingParentStages(stage).sortBy(_.id)
4. if(missing ==Nil){
5. submitMissingTasks(stage)
6. running += stage
7. }else{
8. for(parent <- missing){
9. submitStage(parent)
10. }
11. waiting += stage
12. }
13. }
14. }

对于新提交的job，finalStage的parent stage还未获得，因此submitStage会调用getMissingParentStages()来获得依赖关系：

1. privatedef getMissingParentStages(stage:Stage):List[Stage]={
2. val missing =newHashSet[Stage]
3. val visited =newHashSet[RDD[_]]
4. def visit(rdd: RDD[_]){
5. if(!visited(rdd)){
6. visited += rdd
7. if(getCacheLocs(rdd).contains(Nil)){
8. for(dep <- rdd.dependencies){
9. dep match {
10. case shufDep:ShuffleDependency[_,_]=>
11. val mapStage = getShuffleMapStage(shufDep, stage.priority)
12. if(!mapStage.isAvailable){
13. missing += mapStage
14. }
15. case narrowDep:NarrowDependency[_]=>
16. visit(narrowDep.rdd)
17. }
18. }
19. }
20. }
21. }
22. visit(stage.rdd)
23. missing.toList
24. }

这里parent stage是通过RDD的依赖关系递归遍历获得。对于Wide Dependecy也就是Shuffle Dependecy，Spark会产生新的mapStage作为finalStage的parent，而对于Narrow Dependecy Spark则不会产生新的stage。这里对stage的划分是按照上面提到的作为划分依据的，因此对于本段开头提到的两种job，第一种job只会产生一个finalStage，而第二种job会产生finalStage和mapStage。

当stage DAG产生以后，针对每个stage需要产生task去执行，故在这会调用submitMissingTasks()：

1. privatedef submitMissingTasks(stage:Stage){
2. val myPending = pendingTasks.getOrElseUpdate(stage,newHashSet)
3. myPending.clear()
4. var tasks =ArrayBuffer[Task[_]]()
5. if(stage.isShuffleMap){
6. for(p <-0until stage.numPartitions if stage.outputLocs(p)==Nil){
7. val locs = getPreferredLocs(stage.rdd, p)
8. tasks +=newShuffleMapTask(stage.id, stage.rdd, stage.shuffleDep.get, p, locs)
9. }
10. }else{
11. val job = resultStageToJob(stage)
12. for(id <-0until job.numPartitions if(!job.finished(id))){
13. val partition = job.partitions(id)
14. val locs = getPreferredLocs(stage.rdd, partition)
15. tasks +=newResultTask(stage.id, stage.rdd, job.func, partition, locs, id)
16. }
17. }
18. if(tasks.size >0){
19. myPending ++= tasks
20. taskSched.submitTasks(
21. newTaskSet(tasks.toArray, stage.id, stage.newAttemptId(), stage.priority))
22. if(!stage.submissionTime.isDefined){
23. stage.submissionTime =Some(System.currentTimeMillis())
24. }
25. }else{
26. running -= stage
27. }
28. }

首先根据stage所依赖的RDD的partition的分布，会产生出与partition数量相等的task，这些task根据partition的locality进行分布；其次对于finalStage或是mapStage会产生不同的task；最后所有的task会封装到TaskSet内提交到TaskScheduler去执行。

至此job在DAGScheduler内的启动过程全部完成，交由TaskScheduler执行task，当task执行完后会将结果返回给DAGScheduler，DAGScheduler调用handleTaskComplete()处理task返回:

1. privatedef handleTaskCompletion(event:CompletionEvent){
2. val task =event.task
3. val stage = idToStage(task.stageId)
5. def markStageAsFinished(stage:Stage)={
6. val serviceTime = stage.submissionTime match {
7. caseSome(t)=>"%.03f".format((System.currentTimeMillis()- t)/1000.0)
8. case _ =>"Unkown"
9. }
10. logInfo("%s (%s) finished in %s s".format(stage, stage.origin, serviceTime))
11. running -= stage
12. }
13. event.reason match {
14. caseSuccess=>
15. ...
16. task match {
17. case rt:ResultTask[_, _]=>
18. ...
19. case smt:ShuffleMapTask=>
20. ...
21. }
22. caseResubmitted=>
23. ...
25. caseFetchFailed(bmAddress, shuffleId, mapId, reduceId)=>
26. ...
27. case other =>
28. abortStage(idToStage(task.stageId), task +" failed: "+ other)
29. }
30. }

每个执行完成的task都会将结果返回给DAGScheduler，DAGScheduler根据返回结果来进行进一步的动作。

RDD的计算

RDD的计算是在task中完成的。我们之前提到task分为ResultTask和ShuffleMapTask，我们分别来看一下这两种task具体的执行过程。

• ResultTask

  1. overridedef run(attemptId:Long): U ={
  2. val context =newTaskContext(stageId, partition, attemptId)
  3. try{
  4. func(context, rdd.iterator(split, context))
  5. }finally{
  6. context.executeOnCompleteCallbacks()
  7. }
  8. }

• ShuffleMapTask

  1. overridedef run(attemptId:Long):MapStatus={
  2. val numOutputSplits = dep.partitioner.numPartitions
  4. val taskContext =newTaskContext(stageId, partition, attemptId)
  5. try{
  6. val buckets =Array.fill(numOutputSplits)(newArrayBuffer[(Any,Any)])
  7. for(elem <- rdd.iterator(split, taskContext)){
  8. val pair = elem.asInstanceOf[(Any,Any)]
  9. val bucketId = dep.partitioner.getPartition(pair._1)
  10. buckets(bucketId)+= pair
  11. }
  13. val compressedSizes =newArray[Byte](numOutputSplits)
  15. val blockManager =SparkEnv.get.blockManager
  16. for(i <-0until numOutputSplits){
  17. val blockId ="shuffle_"+ dep.shuffleId +"_"+ partition +"_"+ i
  18. val iter:Iterator[(Any,Any)]= buckets(i).iterator
  19. val size = blockManager.put(blockId, iter,StorageLevel.DISK_ONLY,false)
  20. compressedSizes(i)=MapOutputTracker.compressSize(size)
  21. }
  23. returnnewMapStatus(blockManager.blockManagerId, compressedSizes)
  24. }finally{
  25. taskContext.executeOnCompleteCallbacks()
  26. }
  27. }

ResultTask和ShuffleMapTask都会调用RDD的iterator()来计算和转换RDD，不同的是：ResultTask转换完RDD后调用func()计算结果；而ShufflerMapTask则将其放入blockManager中用来shuffle。

RDD的计算调用iterator()，iterator()在内部调用compute()从RDD依赖关系的根开始计算：

1. finaldef iterator(split:Partition, context:TaskContext):Iterator[T]={
2. if(storageLevel !=StorageLevel.NONE){
3. SparkEnv.get.cacheManager.getOrCompute(this, split, context, storageLevel)
4. }else{
5. computeOrReadCheckpoint(split, context)
6. }
7. }
9. private[spark]def computeOrReadCheckpoint(split:Partition, context:TaskContext):Iterator[T]={
10. if(isCheckpointed){
11. firstParent[T].iterator(split, context)
12. }else{
13. compute(split, context)
14. }
15. }

至此大致分析了TaskSchedulerListener，包括DAGScheduler内部的结构，job生命周期内的活动，RDD是何时何地计算的。接下来我们分析一下task在TaskScheduler内干了什么。

TaskScheduler

前面也提到了Spark实现了三种不同的TaskScheduler，包括LocalSheduler、ClusterScheduler和MesosScheduler。LocalSheduler是一个在本地执行的线程池，DAGScheduler提交的所有task会在线程池中被执行，并将结果返回给DAGScheduler。MesosScheduler依赖于Mesos进行调度，笔者对Mesos了解甚少，因此不做分析。故此章节主要分析ClusterScheduler模块。

ClusterScheduler模块与deploy模块和executor模块耦合较为紧密，因此在分析ClUsterScheduler时也会顺带介绍deploy和executor模块。

首先我们来看一下ClusterScheduler的类图：

ClusterScheduler的启动会伴随SparkDeploySchedulerBackend的启动，而backend会将自己分为两个角色：首先是driver，driver是一个local运行的actor，负责与remote的executor进行通行，提交任务，控制executor；其次是StandaloneExecutorBackend，Spark会在每一个slave node上启动一个StandaloneExecutorBackend进程，负责执行任务，返回执行结果。

ClusterScheduler的启动

在SparkContext实例化的过程中，ClusterScheduler被随之实例化，同时赋予其SparkDeploySchedulerBackend：

1. master match {
2. ...
4. case SPARK_REGEX(sparkUrl)=>
5. val scheduler =newClusterScheduler(this)
6. val backend =newSparkDeploySchedulerBackend(scheduler,this, sparkUrl, appName)
7. scheduler.initialize(backend)
8. scheduler
10. case LOCAL_CLUSTER_REGEX(numSlaves, coresPerSlave, memoryPerSlave)=>
11. ...
12. case _ =>
13. ...
14. }
15. }
16. taskScheduler.start()

ClusterScheduler的启动会启动SparkDeploySchedulerBackend，同时启动daemon进程来检查speculative task：

1. overridedef start(){
2. backend.start()
4. if(System.getProperty("spark.speculation","false")=="true"){
5. newThread("ClusterScheduler speculation check"){
6. setDaemon(true)
8. overridedef run(){
9. while(true){
10. try{
11. Thread.sleep(SPECULATION_INTERVAL)
12. }catch{
13. case e:InterruptedException=>{}
14. }
15. checkSpeculatableTasks()
16. }
17. }
18. }.start()
19. }
20. }

SparkDeploySchedulerBacked的启动首先会调用父类的start()，接着它会启动client，并由client连接到master向每一个node的worker发送请求启动StandaloneExecutorBackend。这里的client、master、worker涉及到了deploy模块，暂时不做具体介绍。而StandaloneExecutorBackend则涉及到了executor模块，它主要的功能是在每一个node创建task可以运行的环境，并让task在其环境中运行。

1. overridedef start(){
2. super.start()
4. val driverUrl ="akka://spark@%s:%s/user/%s".format(
5. System.getProperty("spark.driver.host"),System.getProperty("spark.driver.port"),
6. StandaloneSchedulerBackend.ACTOR_NAME)
7. val args =Seq(driverUrl,"","","")
8. val command =Command("spark.executor.StandaloneExecutorBackend", args, sc.executorEnvs)
9. val sparkHome = sc.getSparkHome().getOrElse(
10. thrownewIllegalArgumentException("must supply spark home for spark standalone"))
11. val appDesc =newApplicationDescription(appName, maxCores, executorMemory, command, sparkHome)
13. client =newClient(sc.env.actorSystem, master, appDesc,this)
14. client.start()
15. }

在StandaloneSchedulerBackend中会创建DriverActor，它就是local的driver，以actor的方式与remote的executor进行通信。

1. overridedef start(){
2. val properties =newArrayBuffer[(String,String)]
3. val iterator =System.getProperties.entrySet.iterator
4. while(iterator.hasNext){
5. val entry = iterator.next
6. val (key, value)=(entry.getKey.toString, entry.getValue.toString)
7. if(key.startsWith("spark.")){
8. properties +=((key, value))
9. }
10. }
11. driverActor = actorSystem.actorOf(
12. Props(newDriverActor(properties)), name =StandaloneSchedulerBackend.ACTOR_NAME)
13. }

在client实例化之前，会将StandaloneExecutorBackend的启动环境作为参数传递给client，而client启动时会将此提交给master，由master分发给所有node上的worker，worker会配置环境并创建进程启动StandaloneExecutorBackend。

至此ClusterScheduler的启动，local driver的创建，remote executor环境的启动所有过程都已结束，ClusterScheduler等待DAGScheduler提交任务。

ClusterScheduler提交任务

DAGScheduler会调用ClusterScheduler提交任务，任务会被包装成TaskSetManager并等待调度：

1. overridedef submitTasks(taskSet:TaskSet){
2. val tasks = taskSet.tasks
3. logInfo("Adding task set "+ taskSet.id +" with "+ tasks.length +" tasks")
4. this.synchronized{
5. val manager =newTaskSetManager(this, taskSet)
6. activeTaskSets(taskSet.id)= manager
7. activeTaskSetsQueue += manager
8. taskSetTaskIds(taskSet.id)=newHashSet[Long]()
10. if(hasReceivedTask ==false){
11. starvationTimer.scheduleAtFixedRate(newTimerTask(){
12. overridedef run(){
13. if(!hasLaunchedTask){
14. logWarning("Initial job has not accepted any resources; "+
15. "check your cluster UI to ensure that workers are registered")
16. }else{
17. this.cancel()
18. }
19. }
20. }, STARVATION_TIMEOUT, STARVATION_TIMEOUT)
21. }
22. hasReceivedTask =true;
23. }
24. backend.reviveOffers()
25. }

在任务提交的同时会启动定时器，如果任务还未被执行，定时器持续发出警告直到任务被执行。同时会调用StandaloneSchedulerBackend的reviveOffers()，而它则会通过actor向driver发送ReviveOffers，driver收到ReviveOffers后调用makeOffers()：

1. // Make fake resource offers on just one executor
2. def makeOffers(executorId:String){
3. launchTasks(scheduler.resourceOffers(
4. Seq(newWorkerOffer(executorId, executorHost(executorId), freeCores(executorId)))))
5. }
7. // Launch tasks returned by a set of resource offers
8. def launchTasks(tasks:Seq[Seq[TaskDescription]]){
9. for(task <- tasks.flatten){
10. freeCores(task.executorId)-=1
11. executorActor(task.executorId)!LaunchTask(task)
12. }
13. }

makeOffers()会向ClusterScheduler申请资源，并向executor提交LauchTask请求。

接下来LaunchTask会进入executor模块，StandaloneExecutorBackend在收到LaunchTask请求后会调用Executor执行task:

1. overridedef receive ={
2. caseRegisteredExecutor(sparkProperties)=>
3. ...
4. caseRegisterExecutorFailed(message)=>
5. ...
6. caseLaunchTask(taskDesc)=>
7. logInfo("Got assigned task "+ taskDesc.taskId)
8. executor.launchTask(this, taskDesc.taskId, taskDesc.serializedTask)
10. caseTerminated(_)|RemoteClientDisconnected(_, _)|RemoteClientShutdown(_, _)=>
11. ...
12. }
14. def launchTask(context:ExecutorBackend, taskId:Long, serializedTask:ByteBuffer){
15. threadPool.execute(newTaskRunner(context, taskId, serializedTask))
16. }

Executor内部是一个线程池，每一个提交的task都会包装为TaskRunner交由threadpool执行：

1. classTaskRunner(context:ExecutorBackend, taskId:Long, serializedTask:ByteBuffer)
2. extendsRunnable{
4. overridedef run(){
5. SparkEnv.set(env)
6. Thread.currentThread.setContextClassLoader(urlClassLoader)
7. val ser =SparkEnv.get.closureSerializer.newInstance()
8. logInfo("Running task ID "+ taskId)
9. context.statusUpdate(taskId,TaskState.RUNNING, EMPTY_BYTE_BUFFER)
10. try{
11. SparkEnv.set(env)
12. Accumulators.clear()
13. val (taskFiles, taskJars, taskBytes)=Task.deserializeWithDependencies(serializedTask)
14. updateDependencies(taskFiles, taskJars)
15. val task = ser.deserialize[Task[Any]](taskBytes,Thread.currentThread.getContextClassLoader)
16. logInfo("Its generation is "+ task.generation)
17. env.mapOutputTracker.updateGeneration(task.generation)
18. val value = task.run(taskId.toInt)
19. val accumUpdates =Accumulators.values
20. val result =newTaskResult(value, accumUpdates)
21. val serializedResult = ser.serialize(result)
22. logInfo("Serialized size of result for "+ taskId +" is "+ serializedResult.limit)
23. context.statusUpdate(taskId,TaskState.FINISHED, serializedResult)
24. logInfo("Finished task ID "+ taskId)
25. }catch{
26. case ffe:FetchFailedException=>{
27. val reason = ffe.toTaskEndReason
28. context.statusUpdate(taskId,TaskState.FAILED, ser.serialize(reason))
29. }
31. case t:Throwable=>{
32. val reason =ExceptionFailure(t)
33. context.statusUpdate(taskId,TaskState.FAILED, ser.serialize(reason))
35. // TODO: Should we exit the whole executor here? On the one hand, the failed task may
36. // have left some weird state around depending on when the exception was thrown, but on
37. // the other hand, maybe we could detect that when future tasks fail and exit then.
38. logError("Exception in task ID "+ taskId, t)
39. //System.exit(1)
40. }
41. }
42. }
43. }

其中task.run()则真正执行了task中的任务，如前RDD的计算章节所述。返回值被包装成TaskResult返回。

至此task在ClusterScheduler内运行的流程有了一个大致的介绍，当然这里略掉了许多异常处理的分支，但这不影响我们对主线的了解

http://jerryshao.me/architecture/2013/04/21/Spark%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90%E4%B9%8B-scheduler%E6%A8%A1%E5%9D%97/