设为首页 加入收藏

TOP

Spark一级资源调度Shedule机制及SpreadOut模式源码深入剖析
2019-03-14 01:21:25 】 浏览:58
Tags:Spark 一级 资源 调度 Shedule 机制 SpreadOut 模式 源码 深入 剖析

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

1 Shedule 在哪里?干什么?

Shedule()发生在Master中,那么Master都负责什么呢?可以看到只要发生以下任何事件,就会重新执行Shedule()

  • RegisterWorker
  • RegisterApplication
  • ExecutorStateChanged
  • RequestSubmitDriver
  • completeRecovery
  • relaunchDriver
  • removeApplication
  • handleRequestExecutors
  • handleKillExecutors
  • removeDriver
因此,可以看出所谓的一级资源调度,在Local-cluster部署模式和Standalone部署模式中,其实就是基于Master实现的资源调度,更确切的说是对Driver的资源调度和对Application(参数指定数量的Executor)的资源调度。

2 Master的天子王权(除Yarn资源和K8s容器编排)

Master是Local-cluster部署模式和Standalone部署模式中,最为核心的管理组件。Master将直接决定整个集群的可用性,容错性,可用性。可谓位于整个Spark集群中最重要,最核心的位置。职责如下:

  • Worker的管理
  • Application的管理
  • Driver的管理
  • 统一管理和分配集群中的资源(如内存和cpu)
  • 接收各个Worker的注册,状态更新,心跳
  • Driver和Application的注册

3 Driver 的前世今生?是什么?如何纳管?

  • Driver的诞生来源于Master接收到RequestSubmitDriver请求,那么RequestSubmitDriver来源于何处,这又要从SparkSubmit类说起,先上代码段,看看STANDALONE_CLUSTER_SUBMIT_CLASS,就从这里开始:

        private[deploy] val YARN_CLUSTER_SUBMIT_CLASS = "org.apache.spark.deploy.yarn.YarnClusterApplication"   
        private[deploy] val REST_CLUSTER_SUBMIT_CLASS = classOf[RestSubmissionClientApp].getName()
        private[deploy] val STANDALONE_CLUSTER_SUBMIT_CLASS = classOf[ClientApp].getName()
        private[deploy] val KUBERNETES_CLUSTER_SUBMIT_CLASS ="org.apache.spark.deploy.k8s.submit.KubernetesClientApplication"
    复制代码
  • 这里开始封装Spark-submit提交的各个参数,同时在StandAlone模式下,我们开始关注ClientEndpoint它是一个终端.

      // In standalone cluster mode, use the REST client to submit the application (Spark 1.3+).
      // All Spark parameters are expected to be passed to the client through system properties.
      if (args.isStandaloneCluster) {
        if (args.useRest) {
          childMainClass = REST_CLUSTER_SUBMIT_CLASS
          childArgs += (args.primaryResource, args.mainClass)
        } else {
        
        
          // In legacy standalone cluster mode, use Client as a wrapper around the user class
          childMainClass = STANDALONE_CLUSTER_SUBMIT_CLASS   <= 神来之笔ClientApp
          
          
          if (args.supervise) { childArgs += "--supervise" }
          Option(args.driverMemory).foreach { m => childArgs += ("--memory", m) }
          Option(args.driverCores).foreach { c => childArgs += ("--cores", c) }
          childArgs += "launch"
          childArgs += (args.master, args.primaryResource, args.mainClass)
        }
        if (args.childArgs != null) {
          childArgs ++= args.childArgs
        }
      }
    复制代码
  • ClientApp (ClientEndpoint) 开始向Master异步发送RequestSubmitDriver请求,也就是说:一次Spark-Submit提交,就会发送一次RequestSubmitDriver请求,进而生成一个资源申请的Driver。

     val driverDescription = new DriverDescription(
           driverArgs.jarUrl,
           driverArgs.memory,
           driverArgs.cores,
           driverArgs.supervise,
           command)
         asyncSendToMasterAndForwardReply[SubmitDriverResponse](
           RequestSubmitDriver(driverDescription))
    复制代码
  • Master接收到提交的资源申请,开始向自己的成员变量drivers中放入一个Driver,也即每一次任务提交的的资源申请驱动。

       case RequestSubmitDriver(description) =>
            if (state != RecoveryState.ALIVE) {
              val msg = s"${Utils.BACKUP_STANDALONE_MASTER_PREFIX}: $state. " +
                "Can only accept driver submissions in ALIVE state."
              context.reply(SubmitDriverResponse(self, false, None, msg))
            } else {
              logInfo("Driver submitted " + description.command.mainClass)
              val driver = createDriver(description)
              persistenceEngine.addDriver(driver)
              waitingDrivers += driver
              drivers.add(driver)
              schedule()
      
              // TODO: It might be good to instead have the submission client poll the master to determine
              //       the current status of the driver. For now it's simply "fire and forget".
      
              context.reply(SubmitDriverResponse(self, true, Some(driver.id),
                s"Driver successfully submitted as ${driver.id}"))
            }
    复制代码
  • 封装资源申请实体DriverInfo

    private def createDriver(desc: DriverDescription): DriverInfo = {
      val now = System.currentTimeMillis()
      val date = new Date(now)
      new DriverInfo(now, newDriverId(date), desc, date)
    }
    复制代码
总结:Master的receiveAndReply方法接收ClientEndpoint发送的消息RequestSubmitDriver,将收到的Driver注册到waitingDrivers。基于此,才会有后面的基于Driver的一级资源调度。
  RequestSubmitDriver详情请参考这篇博客,比我的更详细。https://blog.csdn.net/u011564172/article/details/68496848
复制代码

4 Application 的前世今生?和Driver渊源?如何纳管?

  • 差一点就疯了,Application和Driver完全不是一个概念。Driver的诞生发生在Spark-submit阶段。而Application的诞生发生在DAG调度阶段,也即SparkContext实例化阶段。拼了非讲清不可。

  • Master 最终会根据Application的资源申请,把appDesc放入apps队列中,并对Application进行资源调度。

      val appDesc = ApplicationDescription(sc.appName, maxCores, sc.executorMemory, command,
           webUrl, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor, initialExecutorLimit)
          
      SparkContext:
      
      SparkContext -> StandaloneSchedulerBackend -> StandaloneAppClient.start() 
      -> registerWithMaster -> masterRef.send(RegisterApplication(appDescription, self)) -> 
      
      Master:
      
      -> apps += app ->
      shedule()[Driver启动后,调用startExecutorsOnWorkers()->allocateWorkerResourceToExecutors]
    复制代码
  • Master端点registerApplication

    private def registerApplication(app: ApplicationInfo): Unit = { val appAddress = app.driver.address if (addressToApp.contains(appAddress)) { logInfo("Attempted to re-register application at same address: " + appAddress) return }

      applicationMetricsSystem.registerSource(app.appSource)
      apps += app
      idToApp(app.id) = app
      endpointToApp(app.driver) = app
      addressToApp(appAddress) = app
      waitingApps += app
    }
    复制代码

5 Master的职责再讲

  • 首先集群启动之后,Worker会向Master注册,同时携带身份标识和资源情况(如ID,host,port,cpu核数,内存大小),那么这些资源交由Master纳管后,Master会按照一定的资源调度策略分配给Driver和Application。
  • Master给Driver分配完资源后,将会向Worker发送启动Driver命令,Worker接收到命令后,开始启动Driver。
  • Master给Application分配完资源后,将向Worker发送启动Executor命令,Worker接收到命令后,开始启动Executor。

6 Shedule()神秘面纱

6.1 Shedule 核心思想

  • 打乱洗牌存活的Worker,在Worker资源满足的情况下,启动Executor。

  • 神来之笔(Driver资源调度)==> launchDriver(worker, driver)

  • 神来之笔(Executor调度)==> startExecutorsOnWorkers()

        * Schedule the currently available resources among waiting apps. This method will be called
        * every time a new app joins or resource availability changes.
        
        private def schedule(): Unit = {
          if (state != RecoveryState.ALIVE) {
            return
          }
          // Drivers take strict precedence over executors
          val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(_.state == WorkerState.ALIVE))
          val numWorkersAlive = shuffledAliveWorkers.size
          var curPos = 0
          for (driver <- waitingDrivers.toList) { // iterate over a copy of waitingDrivers
            // We assign workers to each waiting driver in a round-robin fashion. For each driver, we
            // start from the last worker that was assigned a driver, and continue onwards until we have
            // explored all alive workers.
            var launched = false
            var numWorkersVisited = 0
            while (numWorkersVisited < numWorkersAlive && !launched) {
              val worker = shuffledAliveWorkers(curPos)
              numWorkersVisited += 1
              if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
              
                launchDriver(worker, driver)      <= 神来之笔(Driver资源调度)
                
                waitingDrivers -= driver
                launched = true
              }
              curPos = (curPos + 1) % numWorkersAlive
            }
          }
          
          
          startExecutorsOnWorkers()               <= 神来之笔(Executor调度)
          
        }
    复制代码

6.2 launchDriver 挖一挖

  • 发送到Worker开始启动driver

    private def launchDriver(worker: WorkerInfo, driver: DriverInfo) {
      logInfo("Launching driver " + driver.id + " on worker " + worker.id)
      worker.addDriver(driver)
      driver.worker = Some(worker)
      worker.endpoint.send(LaunchDriver(driver.id, driver.desc))
      driver.state = DriverState.RUNNING
    }
    复制代码
  • Worker端的回馈

       case LaunchDriver(driverId, driverDesc) =>
            logInfo(s"Asked to launch driver $driverId")
            val driver = new DriverRunner(
              conf,
              driverId,
              workDir,
              sparkHome,
              driverDesc.copy(command = Worker.maybeUpdateSSLSettings(driverDesc.command, conf)),
              self,
              workerUri,
              securityMgr)
            drivers(driverId) = driver
            driver.start()
      
            coresUsed += driverDesc.cores
            memoryUsed += driverDesc.mem
    复制代码

6.3 startExecutorsOnWorkers 钻一钻

  • coresPerExecutor:参数设置的每一个Executor所使用的内核数,默认为1。

  • app.desc.memoryPerExecutorMB :参数设置的ExecutorMemory。

  • scheduleExecutorsOnWorkers :返回各个Worker上分配的内核数

  • allocateWorkerResourceToExecutors:

      private def startExecutorsOnWorkers(): Unit = {
          // Right now this is a very simple FIFO scheduler. We keep trying to fit in the first app
          // in the queue, then the second app, etc.
          
          for (app <- waitingApps) {
          
            val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) <= 神来之笔(Worker资源情况判断)
            
            // If the cores left is less than the coresPerExecutor,the cores left will not be allocated
            if (app.coresLeft >= coresPerExecutor) {
              // Filter out workers that don't have enough resources to launch an executor
              
              val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE)
                .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB &&
                  worker.coresFree >= coresPerExecutor)
                .sortBy(_.coresFree).reverse                 <= 神来之笔(Worker资源情况判断)
                
              val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps) <= 神来之笔
      
              // Now that we've decided how many cores to allocate on each worker, let's allocate them
              for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) {
              
                allocateWorkerResourceToExecutors(
                  app, assignedCores(pos), app.desc.coresPerExecutor, usableWorkers(pos)) <= 神来之笔
              }
            }
          }
        }
    复制代码

6.3 scheduleExecutorsOnWorkers 较较真

   * Schedule executors to be launched on the workers.
   * Returns an array containing number of cores assigned to each worker.
   *
   * There are two modes of launching executors. The first attempts to spread out an application's
   * executors on as many workers as possible, while the second does the opposite (i.e. launch them
   * on as few workers as possible). The former is usually better for data locality purposes and is
   * the default.
   *
   * The number of cores assigned to each executor is configurable. When this is explicitly set,
   * multiple executors from the same application may be launched on the same worker if the worker
   * has enough cores and memory. Otherwise, each executor grabs all the cores available on the
   * worker by default, in which case only one executor per application may be launched on each
   * worker during one single schedule iteration.
   * Note that when `spark.executor.cores` is not set, we may still launch multiple executors from
   * the same application on the same worker. Consider appA and appB both have one executor running
   * on worker1, and appA.coresLeft > 0, then appB is finished and release all its cores on worker1,
   * thus for the next schedule iteration, appA launches a new executor that grabs all the free
   * cores on worker1, therefore we get multiple executors from appA running on worker1.
   *
   * It is important to allocate coresPerExecutor on each worker at a time (instead of 1 core
   * at a time). Consider the following example: cluster has 4 workers with 16 cores each.
   * User requests 3 executors (spark.cores.max = 48, spark.executor.cores = 16). If 1 core is
   * allocated at a time, 12 cores from each worker would be assigned to each executor.
   * Since 12 < 16, no executors would launch [SPARK-8881].
复制代码
  • spreadOutApps 决定了Executor的分配是集中的,还是按照顺序分散的。

  • oneExecutorPerWorker :如果没有指定coresPerExecutor,那么就说明每一个Worker上只有一个Executor,否则就是多个

  • assignedCores(pos)是返回的数组,其中freeWorkers就是索引0,1,2。对应的可分配的Cores就会是指定Worker上能够分配的。

  • allocateWorkerResourceToExecutors:就是根据打散后的Worker索引,进行Executor的启动,玄机在于每一个Worker是否需要启动多个Executor

      private def scheduleExecutorsOnWorkers(
           app: ApplicationInfo,
           usableWorkers: Array[WorkerInfo],
           spreadOutApps: Boolean): Array[Int] = {
         val coresPerExecutor = app.desc.coresPerExecutor
         val minCoresPerExecutor = coresPerExecutor.getOrElse(1)
         val oneExecutorPerWorker = coresPerExecutor.isEmpty
         val memoryPerExecutor = app.desc.memoryPerExecutorMB
         val numUsable = usableWorkers.length
         val assignedCores = new Array[Int](numUsable) // Number of cores to give to each worker
         val assignedExecutors = new Array[Int](numUsable) // Number of new executors on each worker
         var coresToAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)
     
         /** Return whether the specified worker can launch an executor for this app. */
         def canLaunchExecutor(pos: Int): Boolean = {
           val keepScheduling = coresToAssign >= minCoresPerExecutor
           val enoughCores = usableWorkers(pos).coresFree - assignedCores(pos) >= minCoresPerExecutor
     
           // If we allow multiple executors per worker, then we can always launch new executors.
           // Otherwise, if there is already an executor on this worker, just give it more cores.
           val launchingNewExecutor = !oneExecutorPerWorker || assignedExecutors(pos) == 0
           if (launchingNewExecutor) {
             val assignedMemory = assignedExecutors(pos) * memoryPerExecutor
             val enoughMemory = usableWorkers(pos).memoryFree - assignedMemory >= memoryPerExecutor
             val underLimit = assignedExecutors.sum + app.executors.size < app.executorLimit
             keepScheduling && enoughCores && enoughMemory && underLimit
           } else {
             // We're adding cores to an existing executor, so no need
             // to check memory and executor limits
             keepScheduling && enoughCores
           }
         }
     
         // Keep launching executors until no more workers can accommodate any
         // more executors, or if we have reached this application's limits
         var freeWorkers = (0 until numUsable).filter(canLaunchExecutor)
         while (freeWorkers.nonEmpty) {
           freeWorkers.foreach { pos =>
             var keepScheduling = true
             while (keepScheduling && canLaunchExecutor(pos)) {
               coresToAssign -= minCoresPerExecutor
               assignedCores(pos) += minCoresPerExecutor
     
               // If we are launching one executor per worker, then every iteration assigns 1 core
               // to the executor. Otherwise, every iteration assigns cores to a new executor.
               if (oneExecutorPerWorker) {
                 assignedExecutors(pos) = 1
               } else {
                 assignedExecutors(pos) += 1
               }
     
               // Spreading out an application means spreading out its executors across as
               // many workers as possible. If we are not spreading out, then we should keep
               // scheduling executors on this worker until we use all of its resources.
               // Otherwise, just move on to the next worker.
               if (spreadOutApps) {
                 keepScheduling = false
               }
             }
           }
           freeWorkers = freeWorkers.filter(canLaunchExecutor)
         }
         assignedCores
       }
    复制代码

6.3 allocateWorkerResourceToExecutors 探究竟

  • 通知Worker根据Application的要求,也即根据应用提交时的要求,开始启动Executor。

      private def allocateWorkerResourceToExecutors(
            app: ApplicationInfo,
            assignedCores: Int,
            coresPerExecutor: Option[Int],
            worker: WorkerInfo): Unit = {
          // If the number of cores per executor is specified, we divide the cores assigned
          // to this worker evenly among the executors with no remainder.
          // Otherwise, we launch a single executor that grabs all the assignedCores on this worker.
          val numExecutors = coresPerExecutor.map { assignedCores / _ }.getOrElse(1)
          val coresToAssign = coresPerExecutor.getOrElse(assignedCores)
          for (i <- 1 to numExecutors) {
            val exec = app.addExecutor(worker, coresToAssign)
            launchExecutor(worker, exec)
            app.state = ApplicationState.RUNNING
          }
    }
    复制代码
  • Master 终端发送 LaunchExecutor

      private def launchExecutor(worker: WorkerInfo, exec: ExecutorDesc): Unit = {
          logInfo("Launching executor " + exec.fullId + " on worker " + worker.id)
          worker.addExecutor(exec)
          worker.endpoint.send(LaunchExecutor(masterUrl,
            exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory))
          exec.application.driver.send(
            ExecutorAdded(exec.id, worker.id, worker.hostPort, exec.cores, exec.memory))
        }
    复制代码

7 总结

至此,一级资源调度Shedule机制剖析完毕,真的是剖析的体无完肤啊。贴一张图,该休息了。因为已经0:18了。

秦凯新 于深圳 香港太平山全景 人定胜天

再见 2018 11 12


编程开发网
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇Spark内置框架rpc通讯机制及RpcEn.. 下一篇Spark底层多个MessageLoop循环线..

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

array(4) { ["type"]=> int(8) ["message"]=> string(24) "Undefined variable: jobs" ["file"]=> string(32) "/mnt/wp/cppentry/do/bencandy.php" ["line"]=> int(217) }