阅读 239

Spark二级调度系统Stage划分算法和最佳任务调度细节剖析-Spark商业环境实战

本套系列博客从真实商业环境抽取案例进行总结和分享,并给出Spark源码解读及商业实战指导,请持续关注本套博客。版权声明:本套Spark源码解读及商业实战归作者(秦凯新)所有,禁止转载,欢迎学习。

Spark商业环境实战及调优进阶系列

1. Spark调度系统的组件关系

1.1 Spark调度系统的组件(以StandAlone模式)

  • 一级调度:Cluster Manger (YARN模式下为ResourceManger , Standalone 模式下为 Master )负责将资源分配给Application。这里的资源如:cpu核数,内存,磁盘空间等。

  • 二级调度:Application 进一步将资源分配给Application的各个Task。DAG任务调度,延迟调度等。

  • 任务(Task): Task分为ResultTask和ShuffleMapTask两种,每一个Stage会根据未完成的Partion的多少,创建零到多个Task,DAGScheduer最后将每个Stage中的Task以任务集合(TaskSet)的形式提交给TaskScheduler继续处理。

  • TaskSchedulerImpl :手握StandaloneSchedulerBackend(StandaloneAppClient)和 CoarseGrainedSchedulerBackend(DriverEndpoint)两大通讯端点神器。操控了整个集群资源的汇报和资源调度。

  • CoarseGrainedSchedulerBackend:维护了一个executorDataMap,用于实时拿到最新的且活着的executor用于资源分配。

  • StandaloneSchedulerBackend: 持有TaskSchedulerImpl的引用,目前来看,就是为了初始化启动StandaloneAppClient和DriverEndpoint终端,通过接受消息,实际干活的还是TaskSchedulerImpl。

  • StandaloneAppClient:在Standalone模式下StandaloneSchedulerBackend在启动的时候构造AppClient实例并在该实例start的时候启动了ClientEndpoint这个消息循环体。ClientEndpoint在启动的时候会向Master注册当前程序。(Interface allowing applications to speak with a Spark standalone cluster manager.)

    (1) StandaloneSchedulerBackend 与 CoarseGrainedSchedulerBackend 的前世,可以看到在TaskSchedulerImpl和StandaloneSchedulerBackend相互引用并启动:
      SparkContext -> createTaskScheduler -> new TaskSchedulerImpl(sc) ->
      new StandaloneSchedulerBackend(scheduler, sc, masterUrls)-> scheduler.initialize(backend)
      _taskScheduler.start()-> backend.start()
    复制代码
    (2) StandaloneAppClient 与 DriverEndpoint 的今生:
      StandaloneSchedulerBackend->start()
      
      -> super.start() [CoarseGrainedSchedulerBackend]-> 
      createDriverEndpointRef(properties)->  createDriverEndpoint(properties) ->
      
      -> new StandaloneAppClient(sc.env.rpcEnv, masters, appDesc, this, conf).start()
    复制代码
    (3) StandaloneAppClient 新官上任
      onStart -> registerWithMaster -> tryRegisterAllMasters ->  rpcEnv.setupEndpointRef(masterAddress,
      Master.ENDPOINT_NAME) -> masterRef.send(RegisterApplication(appDescription, self))
    复制代码
    (4)StandaloneAppClient负责的功能如下:
          RegisteredApplication(启动时向Master注册)
          ApplicationRemoved
          ExecutorAdded
          ExecutorUpdated
          WorkerRemoved
          MasterChanged
    复制代码
  • DriverEndpoint:而StandaloneSchedulerBackend的父类CoarseGrainedSchedulerBackend在start的时候会实例化类型为DriverEndpoint(这就是我们程序运行时候的经典对象的Driver)的消息循环体,StandaloneSchedulerBackend专门负责收集Worker上的资源信息, 当Worker端的ExecutorBackend启动的时候会发送RegisteredExecutor信息向DriverEndpoint注册,

    此时StandaloneSchedulerBackend就掌握了当前应用程序拥有的计算资源。TaskSchedulerImpl就是通过StandaloneSchedulerBackend拥有的计算资源来具体运行Task。负责的功能如下:
      StatusUpdate
      ReviveOffers -->
      KillTask
      KillExecutorsOnHost
      RemoveExecutor
      RegisterExecutor()
      StopDriver
      StopExecutors
      RemoveWorker
    复制代码

Spark调度系统总体规律:在Standalone模式下StandaloneSchedulerBackend在启动的时候构造AppClient实例并在该实例start的时候启动了ClientEndpoint这个消息循环体。ClientEndpoint在启动的时候会向Master注册当前程序。而StandaloneSchedulerBackend的父类CoarseGrainedSchedulerBackend在start的时候会实例化类型为DriverEndpoint(这就是我们程序运行时候的经典对象的Driver)的消息循环体,StandaloneSchedulerBackend专门负责收集Worker上的资源信息,当ExecutorBackend启动的时候会发送RegisteredExecutor信息向DriverEndpoint注册,此时StandaloneSchedulerBackend就掌握了当前应用程序拥有的计算资源,TaskScheduler就是通过StandaloneSchedulerBackend拥有的计算资源来具体运行Task。

2. DAGScheduler 核心调度系统

2.1 DAGScheduler 核心成员

  • TaskSchdulerImpl

  • LiveListenerBus

  • MapoutTrackerMaster

  • BlockManagerMaster

  • SparkEnv

  • cacheLocas:缓存每个RDD的所有分区的位置信息,最终建立分区号和位置信息序列映射。为什么是位置序列? 这里着重讲解一下:每一个分区可能存在多个副本机制,因此RDD的每一个分区的BLock可能存在多个节点的BlockManager上,因此是序列。听懂了吗??

        new HashMap[Int, IndexedSeq[Seq[TaskLocation]]]
    复制代码
  • MessageScheduler:职责是对失败的Stage进行重试,如下面的执行线程代码段:

    private val messageScheduler =
     ThreadUtils.newDaemonSingleThreadScheduledExecutor("dag-scheduler-message")
    
      case FetchFailed -> messageScheduler.schedule(new Runnable {
          override def run(): Unit = eventProcessLoop.post(ResubmitFailedStages)
        }, DAGScheduler.RESUBMIT_TIMEOUT, TimeUnit.MILLISECONDS)
      }
    复制代码
  • getPreferredLocs:重量级方法,分区最大偏好位置获取。最终把分区最佳偏好位置序列放在cacheLocas中,获取不到,调用rdd.preferredLocations方法获取。

      getPreferredLocs 
      
      -> getPreferredLocsInternal 
      
      -> getCacheLocs(rdd)(partition) ---> cacheLocs(rdd.id) = locs() -> 返回 cached
                 (取不到直接放进内存后,再返回偏好序列)
      
      -> val rddPrefs = rdd.preferredLocations(rdd.partitions(partition)).toList
    复制代码

    getCacheLocs 方法代码段:

      def getCacheLocs(rdd: RDD[_]): IndexedSeq[Seq[TaskLocation]] =         
          cacheLocs.synchronized {
      if (!cacheLocs.contains(rdd.id)) {
      val locs: IndexedSeq[Seq[TaskLocation]] = if (rdd.getStorageLevel == StorageLevel.NONE) {
      IndexedSeq.fill(rdd.partitions.length)(Nil)
      } else {
      val blockIds =
      rdd.partitions.indices.map(index => RDDBlockId(rdd.id, index)).toArray[BlockId]
      blockManagerMaster.getLocations(blockIds).map { bms =>
      bms.map(bm => TaskLocation(bm.host, bm.executorId))
      }
    }
      cacheLocs(rdd.id) = locs
      }
      cacheLocs(rdd.id)
      }
    复制代码
  • eventProccessLoop:大名鼎鼎的DAGSchedulerEventProcessLoop,事件处理线程,负责处理各种事件,如:

      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)
      
      case MapStageSubmitted(jobId, dependency, callSite, listener, properties) =>
      dagScheduler.handleMapStageSubmitted(jobId, dependency, callSite, listener, properties)
    
      case StageCancelled(stageId) =>
      dagScheduler.handleStageCancellation(stageId)
    
      case JobCancelled(jobId) =>
      dagScheduler.handleJobCancellation(jobId)
    
      case JobGroupCancelled(groupId) =>
      dagScheduler.handleJobGroupCancelled(groupId)
    
      case AllJobsCancelled =>
      dagScheduler.doCancelAllJobs()
    
      case ExecutorAdded(execId, host) =>
      dagScheduler.handleExecutorAdded(execId, host)
    
      case ExecutorLost(execId) =>
      dagScheduler.handleExecutorLost(execId, fetchFailed = false)
    
       case BeginEvent(task, taskInfo) =>
       dagScheduler.handleBeginEvent(task, taskInfo)
    
      case GettingResultEvent(taskInfo) =>
      dagScheduler.handleGetTaskResult(taskInfo)
    
      case completion: CompletionEvent =>
      dagScheduler.handleTaskCompletion(completion)
    
      case TaskSetFailed(taskSet, reason, exception) =>
      dagScheduler.handleTaskSetFailed(taskSet, reason, exception)
    
      case ResubmitFailedStages =>
      dagScheduler.resubmitFailedStages()
    复制代码

2.2 DAGScheduler 与 job 的 牵手缘分,一个job的前世今生

(1) DAG的有向无环图切分及任务调度job提交流程(以下非代码堆积,从上往下看逻辑核心流程):
    -> DAGScheduler.runJob->submitJob(rdd, func, partitions, callSite, resultHandler,
       properties)
    
    -> DAGScheduler.eventProcessLoop.post(JobSubmitted(
               jobId, rdd, func2, partitions.toArray, callSite, waiter,
               SerializationUtils.clone(properties))) 
    
    -> DAGSchedulerEventProcessLoop-> case JobSubmitted(jobId, dependency, callSite, listener, properties)->
    
    -> DAGScheduler.handleJobSubmitted(jobId, rdd, func, partitions, callSite, listener, properties)
    
    -> DAGScheduler.createResultStage(finalRDD, func, partitions, jobId, callSite) (广度优先算法,建立无环图)
    
    -> DAGScheduler.submitStage(finalStage)
    
    -> DAGScheduler.submitMissingTasks(stage, jobId.get) (提交最前面的Stage0)
    
    -> taskScheduler.submitTasks(new TaskSet(
       tasks.toArray, stage.id, stage.latestInfo.attemptNumber, jobId, properties))
    
    -> taskScheduler.submitTasks(new TaskSet(tasks.toArray, stage.id, stage.latestInfo.attemptNumber, jobId,
       properties)) 
       
    -> createTaskSetManager(taskSet, maxTaskFailures) (TaskSchedulerImpl.submitTasks方法内)
    
    -> taskScheduler ->initialize -> FIFOSchedulableBuilder ->buildPools (TaskSchedulerImpl初始化构建)
    
    -> schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
       (TaskSchedulerImpl.submitTasks方法内,  TaskSetManager pool ,遵循FIFO)
    
    -> backend.reviveOffers()(CoarseGrainedSchedulerBackend)
    
    -> driverEndpoint.send(ReviveOffers)
    
    -> DriverEndpoint.receive  -> case ReviveOffers  makeOffers() (makeOffers封装公共调度任务调度方法)
    
    -> makeOffers(获得活着的Executor及executorHost和freeCores) -> workOffers(CoarseGrainedSchedulerBackend内部)
    
    -> TaskSchedulerImpl.resourceOffers(workOffers)(CoarseGrainedSchedulerBackend引用TaskSchedulerImpl)
    
    -> val sortedTaskSets = TaskSchedulerImpl.rootPool.getSortedTaskSetQueue  (TaskSetManager上场)
    
    -> TaskSchedulerImpl.resourceOfferSingleTaskSet(
                  taskSet, currentMaxLocality, shuffledOffers, availableCpus, tasks)t
    
    -> TaskSetManager.resourceOffer(execId, host, maxLocality))
    
    -> TaskSchedulerImpl.getAllowedLocalityLevel(curTime)  (延迟调度,期待我的下一篇精彩讲解)
    
    -> TaskSetManager.dequeueTask(execId, host, allowedLocality)
    
    -> sched.dagScheduler.taskStarted(task, info)
    
    -> new TaskInfo(taskId, index, attemptNum, curTime, execId, host, taskLocality, speculative)
复制代码

2.3 Job 与 Stage 的分分合合,Stage反向驱动与正向提交

反向驱动:长江后浪推前浪,这里我发现一个奇怪的事情,Spark 2.0版本stage的反向驱动算法和Spark 2.3的居然不一样,这里以Spark 2.3为准:

  • DAGScheduler进行submitStage提交后使命就结束了,最终实现submitStage正向提交任务集合即可:

     ->  DAGScheduler. handleJobSubmitted 
     ->  finalStage = createResultStage(finalRDD, func, partitions, jobId, callSite)
        (封装 finalRDD 和func为最后的Stage)
     ->  submitStage(finalStage) -> submitMissingTasks
    复制代码
  • 但是createResultStage反向驱动算法精彩开始了:

    -> createResultStage 
    -> getOrCreateParentStages -> getShuffleDependencies  (获取父依赖)
    -> getOrCreateShuffleMapStage -> getShuffleDependencies  (不断循环,形成递归)
    -> createShuffleMapStage 
    复制代码
  • 以下代码段作用是获取父stage,有的话直接返回,没有就重新根据final RDD父依赖来创建Stage。注意这里会不断递归调用getOrCreateParentStages,最终建立Stage,也因此

private def getOrCreateParentStages(rdd: RDD[_], firstJobId: Int): List[Stage] = {
getShuffleDependencies(rdd).map { shuffleDep =>
  getOrCreateShuffleMapStage(shuffleDep, firstJobId)
}.toList
复制代码

}

  • getOrCreateShuffleMapStage:创建依赖的依赖的所在Stage,有的话会直接获取,没有就优先创建 父Stage,然后执行子Stage.

    private def getOrCreateShuffleMapStage(
    shuffleDep: ShuffleDependency[_, _, _],
    firstJobId: Int): ShuffleMapStage = {
      shuffleIdToMapStage.get(shuffleDep.shuffleId) match {
    case Some(stage) =>
      stage
    
    case None =>
      // Create stages for all missing ancestor shuffle dependencies.
      getMissingAncestorShuffleDependencies(shuffleDep.rdd).foreach { dep =>
        // Even though getMissingAncestorShuffleDependencies only returns shuffle dependencies
        // that were not already in shuffleIdToMapStage, it's possible that by the time we
        // get to a particular dependency in the foreach loop, it's been added to
        // shuffleIdToMapStage by the stage creation process for an earlier dependency. See
        // SPARK-13902 for more information.
        if (!shuffleIdToMapStage.contains(dep.shuffleId)) {
          createShuffleMapStage(dep, firstJobId)
        }
      }
      // Finally, create a stage for the given shuffle dependency.
      createShuffleMapStage(shuffleDep, firstJobId)
    复制代码

    } }

  • createShuffleMapStage:最终落地方法,就是要返回需要的stage,注意阻塞点就在getOrCreateParentStages,从而一直递归到最顶层。

     def createShuffleMapStage(shuffleDep: ShuffleDependency[_, _, _], jobId: Int):      ShuffleMapStage = {
     val rdd = shuffleDep.rdd
     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
     shuffleIdToMapStage(shuffleDep.shuffleId) = stage
     updateJobIdStageIdMaps(jobId, stage)
     if (!mapOutputTracker.containsShuffle(shuffleDep.shuffleId)) {
     mapOutputTracker.registerShuffle(shuffleDep.shuffleId, rdd.partitions.length)
     }
     stage
     }
    复制代码

2.4 秦凯新原创总结:

总流程是:先创建最顶层Satge,慢慢递归执行创建子stage,类似于堆栈模型。

  • 总流程为什么是这样呢?如无环图: G -> F A -> H -> M L->N,因为由createShuffleMapStage做反向驱动,阻塞点在就在方法内的getOrCreateParentStages,因此先把创建(Final Stage的父stage的创建方法 F, A)放在堆栈底部,不断向上存放(F, A以上的父Stage创建方法)依次到堆栈,但却不执行Stage创建,直到最后最顶层Stage创建方法放到堆栈时,在得到rdd的顶级父亲时,开始执行最顶层Stage创建方法,也即createShuffleMapStage开始从阻塞点递归建立依赖关系,注册Shuffle,执行createShuffleMapStage方法体,然后依次从阻塞点递归向下执行。

  • 注意 getOrCreateShuffleMapStage的缓存机制,即shuffleIdToMapStage,实现了依赖的直接获取,不用再重复执行,如F,A的父Stage的获取。

      G  -> F A -> H -> M  L->N
    
      1 先从 F A 的父依赖开始 开始递归构建stage
      
      2 进而开始建立H以上的stagey以及依赖关系。
      
      3 后建立 F A 与 H stage 的关系,注意H Stage是从shuffleIdToMapStage拿到的,
        最后返回F,A stage ,建立 final RDD(G)与F A的依赖关系
      
      4 最终提交submitStage(finalStage)
    
      最终stage构建顺序为: N -> M  L -> H-> F A  -> G  
    复制代码

总结

本节内容是作者投入大量时间优化后的内容,采用最平实的语言来剖析Spark的任务调度,现在时间为凌晨1:22,最后放张图DAGScheduler.handleTaskCompletion,正向提交任务,放飞吧,spark。

作者:秦凯新 20181102

关注下面的标签,发现更多相似文章
评论