[第七章] spark 资源调度算法深入剖析

前面几章我们重点讲述了spark的的原理,sparkContext的初使化,spark主备切换,master的注册等。其中我们分析源码时,不管是driver,还是application在注册Master时都,在最后都有这样的一个方法调用:

 ...
 schedule()

这个方法其实就是资源调度算法(这个方法太重要了,也是核心中的核心)
下面我们就来深入分析这个方法。

 private def schedule() {
  #当master不是alive时,直接reuturn
  #也就是说Standby是不参与资源调度的
    if (state != RecoveryState.ALIVE) { return }
//Random.shuffle作用就是把集合随机打乱
//取出workers中所有之前注册的worker,进行过滤,必须 状态 是Alive的worker
//把worker随机的打乱
 val shuffledAliveWorkers = Random.shuffle(workers.toSeq.filter(
_.state == WorkerState.ALIVE))
    val numWorkersAlive = shuffledAliveWorkers.size

为什么要调度driver,大家想一下什么时候会注册driver,并且导致driver被调度
其实只有在模式是yarn-cluster提交后,才会注册driver,因为standalone与yarn-client
都会在本地启动dirver,而不会来注册driver,就更不可能被master来调度
所以说下面的这个for只会运行在yarn-cluster模式下提交下。

 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中的条件,当还有活着的worker没有被遍历到,就继续遍历
       * 而且这个driver在这个worker中还没有启动,launched=false
       */
      while (numWorkersVisited < numWorkersAlive && !launched) {
        val worker = shuffledAliveWorkers(curPos)
        numWorkersVisited += 1
        /**
         * 如果这个worker的空闲内存容量 大于等于driver所需的内存
         * 而且worker空间的CPU大于等于driver所需的CPU数量
         */
        if (worker.memoryFree >= driver.desc.mem && worker.coresFree >= driver.desc.cores) {
         //启动driver
          launchDriver(worker, driver)
          //把此driver从waitingDrivers中去掉
          waitingDrivers -= driver
          launched = true
        }
        //将指针指向下一个worker
        curPos = (curPos + 1) % numWorkersAlive
      }
    }

上面就是对Drvier的调度。
接着我们看对Application的资源调度
/**
* Application的调度机制(核心之核心 )
* 两种算法:一种是spreadOutApps(默认),另一种是非spreadOutApps
*
* 通过个算法,其实 会将每个application,要启动的executor都平均分配 到每个worker上
* 比如有20cpu core,有10个worker,那么实际会遍历两遍,每次循环,每个worker分配一个core
* 最后每个worker分配了两个core
*/

  if (spreadOutApps) {
          for (app <- waitingApps if app.coresLeft > 0) {
        //从workerk中,过滤出状态是ALIVE的
        val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE)
          .filter(canUse(app, _)).sortBy(_.coresFree).reverse
        val numUsable = usableWorkers.length
        //创建一个空数组,存储了要分配的每个worker的cpu
        val assigned = new Array[Int](numUsable) // Number of cores to give on each node
        //获取到底可以分配多少个cpu,取application所需的cpu数量与worker可用的cpu数量的最小值
        var toAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum)
        var pos = 0
        //只要有还有要分配的cpu没有分配完就while
        while (toAssign > 0) {
          //如果可用的cpu大于已经分配的cpu数量,其实就是还有可用的cpu
          if (usableWorkers(pos).coresFree - assigned(pos) > 0) {
            //将要分配的cpu数量减一
            toAssign -= 1
            //给这个worker分配的cpu加1
            assigned(pos) += 1
          }
          //指定指向下一个worker
          pos = (pos + 1) % numUsable
        }
        // Now that we've decided how many cores to give on each node, let's actually give them
       //给每个worker分配了application需要的cpu core后
        for (pos <- 0 until numUsable) {
          //判断这个worker已经分配了core
          if (assigned(pos) > 0) {
            //创建了ExecutorDesc对象,封装了executor的信息
            //将这个executor添加到application缓存区
            val exec = app.addExecutor(usableWorkers(pos), assigned(pos))
            //在worker上启动Executor
            launchExecutor(usableWorkers(pos), exec)
            //将application的状态为RUNNING
            app.state = ApplicationState.RUNNING
           }
        }
      }
    }

关于这种调度算法总结:

我们之前说了在spark-submit中指定了多少个executor,每个execuotr需要多少个cpu core 实际上基本这个机制,最后,executor的实际数量,每个executor需要的core,可能与配置不一样 因为这里我们是基于总的cpu来分配的,就是说比如,我们配置了需要三个executor来启动application, 每个executor需要三个core,那么就总需9个core,其实在这种算法中,如果我们有9个worker,会给每个 worker分配一个core,然后给每个worker启动一个executor.最后其实是启动了9个executor,每个 executor有一个core

第二种调度算法:

这种算法与上面的正好相反,每个application,都尽可能少的分配到worker上去, 比如总共有10个worker,每个有10个core application总共要分配20个core,那么只会分配到两个worker上,每个worker都占满了这10个core那么其它的application只能分配另外的worker上去了。 所以我们在spark-submit中配置了要10个executor,每个execuotr需要2个core 那么共需要20个core,但这种算法中,其实只会启动两个executor,每个executor有10个core

//遍历worker,并且状态是ALIVE,还有空闲的cpu的worker
      for (worker <- workers if worker.coresFree > 0 && worker.state == WorkerState.ALIVE) {
       //遍历application,并且还有需要分配的core的applicztion
        for (app <- waitingApps if app.coresLeft > 0) {
          //判断当前这个worker可以被application使用
          if (canUse(app, worker)) {
            //取worker可用的cpu数量与application要分配的cpu数量的最小值
            val coresToUse = math.min(worker.coresFree, app.coresLeft)
            //如果小于0,说明没有core可分了
            if (coresToUse > 0) {
              val exec = app.addExecutor(worker, coresToUse)
                      //在worker上启动executor
              launchExecutor(worker, exec)
              //设置application的状态是
              app.state = ApplicationState.RUNNING
            }
          }

其中里面有一个非常重要的方法:

launchExecutor(worker, exec)
def launchExecutor(worker: WorkerInfo, exec: ExecutorDesc) {
    logInfo("Launching executor " + exec.fullId + " on worker " + worker.id)
    //将executor加入worker缓存
    worker.addExecutor(exec)
    //向worker的actor发送LaunchExecutor ,在worker中启动executor
    worker.actor ! LaunchExecutor(masterUrl,
      exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory)
    exec.application.driver ! ExecutorAdded(
      exec.id, worker.id, worker.hostPort, exec.cores, exec.memory)
  }

这个方法里的
worker.actor ! LaunchExecutor(masterUrl,
exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory)
正是向worker发送这个消息(封闭了akka的消息通信) ,在worker端调用这个方法来启动Executor(关于worker如何启动Eexecutor与Application,我们在下一节会详细的剖析)

本章中每一个字(包括源码注解)都是作者敲出来的,你感觉有用,帮点击’喜欢’

    原文作者:cariya
    原文地址: https://www.jianshu.com/p/755778cf64b2
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞