Apache Spark源码走读之19 -- standalone cluster模式下资源的申请与释放

欢迎转载，转载请注明出处，徽沪一郎。

概要

本文主要讲述在standalone cluster部署模式下，Spark Application在整个运行期间，资源(主要是cpu core和内存)的申请与释放。

构成Standalone cluster部署模式的四大组成部件如下图所示，分别为Master, worker, executor和driver，它们各自运行于独立的JVM进程。

从资源管理的角度来说

• Master  掌管整个cluster的资源，主要是指cpu core和memory，但Master自身并不拥有这些资源
• Worker 计算资源的实际贡献者，须向Master汇报自身拥有多少cpu core和memory, 在master的指示下负责启动executor
• Executor 执行真正计算的苦力，由master来决定该进程拥有的core和memory数值
• Driver 资源的实际占用者，Driver会提交一到多个job，每个job在拆分成多个task之后，会分发到各个executor真正的执行

这些内容在standalone cluster模式下的容错性分析中也有所涉及，今天主要讲一下资源在分配之后不同场景下是如何被顺利回收的。

资源上报汇聚过程

standalone cluster下最主要的当然是master，master必须先于worker和driver程序正常启动。

当master顺利启动完毕，可以开始worker的启动工作，worker在启动的时候需要向master发起注册，在注册消息中带有本worker节点的cpu core和内存。

调用顺序如下preStart->registerWithMaster->tryRegisterAllMasters

看一看tryRegisterAllMasters的代码

 def tryRegisterAllMasters() {
    for (masterUrl <- masterUrls) {
      logInfo("Connecting to master " + masterUrl + "...")
      val actor = context.actorSelection(Master.toAkkaUrl(masterUrl))
      actor ! RegisterWorker(workerId, host, port, cores, memory, webUi.boundPort, publicAddress)
    }
  }

我们的疑问是RegisterWorker构造函数所需的参数memory和cores是从哪里获取的呢？

注意一下Worker中的main函数会创建WorkerArguments，

  def main(argStrings: Array[String]) {
    SignalLogger.register(log)
    val args = new WorkerArguments(argStrings)
    val (actorSystem, _) = startSystemAndActor(args.host, args.port, args.webUiPort, args.cores,
      args.memory, args.masters, args.workDir)
    actorSystem.awaitTermination()
  }

memory通过函数inferDefaultMemory获取，而cores通过inferDefaultCores获取。

def inferDefaultCores(): Int = {
    Runtime.getRuntime.availableProcessors()
  }

  def inferDefaultMemory(): Int = {
    val ibmVendor = System.getProperty("java.vendor").contains("IBM")
    var totalMb = 0
    try {
      val bean = ManagementFactory.getOperatingSystemMXBean()
      if (ibmVendor) {
        val beanClass = Class.forName("com.ibm.lang.management.OperatingSystemMXBean")
        val method = beanClass.getDeclaredMethod("getTotalPhysicalMemory")
        totalMb = (method.invoke(bean).asInstanceOf[Long] / 1024 / 1024).toInt
      } else {
        val beanClass = Class.forName("com.sun.management.OperatingSystemMXBean")
        val method = beanClass.getDeclaredMethod("getTotalPhysicalMemorySize")
        totalMb = (method.invoke(bean).asInstanceOf[Long] / 1024 / 1024).toInt
      }
    } catch {
      case e: Exception => {
        totalMb = 2*1024
        System.out.println("Failed to get total physical memory. Using " + totalMb + " MB")
      }
    }
    // Leave out 1 GB for the operating system, but don't return a negative memory size
    math.max(totalMb - 1024, 512)
  }

如果已经在配置文件中为显示指定了每个worker的core和memory，则使用配置文件中的值，具体配置参数为SPARK_WORKER_CORES和SPARK_WORKER_MEMORY

Master在收到RegisterWork消息之后，根据上报的信息为每一个worker创建相应的WorkerInfo.

    case RegisterWorker(id, workerHost, workerPort, cores, memory, workerUiPort, publicAddress) =>
    {
      logInfo("Registering worker %s:%d with %d cores, %s RAM".format(
        workerHost, workerPort, cores, Utils.megabytesToString(memory)))
      if (state == RecoveryState.STANDBY) {
        // ignore, don't send response
      } else if (idToWorker.contains(id)) {
        sender ! RegisterWorkerFailed("Duplicate worker ID")
      } else {
        val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,
          sender, workerUiPort, publicAddress)
        if (registerWorker(worker)) {
          persistenceEngine.addWorker(worker)
          sender ! RegisteredWorker(masterUrl, masterWebUiUrl)
          schedule()
        } else {
          val workerAddress = worker.actor.path.address
          logWarning("Worker registration failed. Attempted to re-register worker at same " +
            "address: " + workerAddress)
          sender ! RegisterWorkerFailed("Attempted to re-register worker at same address: "
            + workerAddress)
        }
      }

资源分配过程

如果在worker注册上来的时候，已经有Driver Application注册上来，那么就需要将原先处于未分配资源状态的driver application启动相应的executor。

WorkerInfo在schedule函数中会被使用到，schedule函数处理逻辑概述如下

1. 查看目前存活的worker中剩余的内存是否能够满足application每个task的最低需求，如果是则将该worker加入到可分配资源的队列
2. 根据分发策略，如果是决定将工作平摊到每个worker，则每次在一个worker上占用一个core，直到所有可分配资源耗尽或已经满足driver的需求
3. 如果分发策略是分发到尽可能少的worker，则一次占用尽worker上的可分配core，直到driver的core需求得到满足
4. 根据步骤2或3的结果在每个worker上添加相应的executor，处理函数是addExecutor

为了叙述简单，现仅列出平摊到各个worker的分配处理过程

      for (worker > workers if worker.coresFree > 0 && worker.state == WorkerState.ALIVE) {
        for (app <- waitingApps if app.coresLeft > 0) {
          if (canUse(app, worker)) {
            val coresToUse = math.min(worker.coresFree, app.coresLeft)
            if (coresToUse > 0) {
              val exec = app.addExecutor(worker, coresToUse)
              launchExecutor(worker, exec)
              app.state = ApplicationState.RUNNING
            }
          }
        }
      }

launchExecutor主要负责两件事情

1. 记录下新添加的executor使用掉的cpu core和内存数目，记录过程发生在worker.addExecutor
2. 向worker发送LaunchExecutor指令

  def launchExecutor(worker: WorkerInfo, exec: ExecutorInfo) {
    logInfo("Launching executor " + exec.fullId + " on worker " + worker.id)
    worker.addExecutor(exec)
    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在收到LaunchExecutor指令后，也会记一笔账，将要使用掉的cpu core和memory从可用资源中减去，然后使用ExecutorRunner来负责生成Executor进程，注意Executor运行于独立的进程。代码如下

case LaunchExecutor(masterUrl, appId, execId, appDesc, cores_, memory_) =>
      if (masterUrl != activeMasterUrl) {
        logWarning("Invalid Master (" + masterUrl + ") attempted to launch executor.")
      } else {
        try {
          logInfo("Asked to launch executor %s/%d for %s".format(appId, execId, appDesc.name))
          val manager = new ExecutorRunner(appId, execId, appDesc, cores_, memory_,
            self, workerId, host,
            appDesc.sparkHome.map(userSparkHome => new File(userSparkHome)).getOrElse(sparkHome),
            workDir, akkaUrl, conf, ExecutorState.RUNNING)
          executors(appId + "/" + execId) = manager
          manager.start()
          coresUsed += cores_
          memoryUsed += memory_
          masterLock.synchronized {
            master ! ExecutorStateChanged(appId, execId, manager.state, None, None)
          }
        } catch {
          case e: Exception => {
            logError("Failed to launch executor %s/%d for %s".format(appId, execId, appDesc.name))
            if (executors.contains(appId + "/" + execId)) {
              executors(appId + "/" + execId).kill()
              executors -= appId + "/" + execId
            }
            masterLock.synchronized {
              master ! ExecutorStateChanged(appId, execId, ExecutorState.FAILED, None, None)
            }
          }
        }
      }

在资源分配过程中需要注意到的是如果有多个Driver Application处于等待状态，资源分配的原则是FIFO，先到先得。

资源回收过程

worker中上报的资源最终被driver application中提交的job task所占用，如果application结束(包括正常和异常退出)，application所占用的资源就应该被顺利回收，即将占用的资源重新归入可分配资源行列。

现在的问题转换成Master和Executor如何知道Driver Application已经退出了呢？

有两种不同的处理方式，一种是先道别后离开，一种是不告而别。现分别阐述。

何为先道别后离开，即driver application显式的通知master和executor，任务已经完成了，我要bye了。应用程序显式的调用SparkContext.stop

  def stop() {
    postApplicationEnd()
    ui.stop()
    // Do this only if not stopped already - best case effort.
    // prevent NPE if stopped more than once.
    val dagSchedulerCopy = dagScheduler
    dagScheduler = null
    if (dagSchedulerCopy != null) {
      metadataCleaner.cancel()
      cleaner.foreach(_.stop())
      dagSchedulerCopy.stop()
      taskScheduler = null
      // TODO: Cache.stop()?
      env.stop()
      SparkEnv.set(null)
      ShuffleMapTask.clearCache()
      ResultTask.clearCache()
      listenerBus.stop()
      eventLogger.foreach(_.stop())
      logInfo("Successfully stopped SparkContext")
    } else {
      logInfo("SparkContext already stopped")
    }
  }

显式调用SparkContext.stop的一个主要功能是会去显式的停止Executor，具体下达StopExecutor指令的代码见于CoarseGrainedSchedulerBackend中的stop函数

  override def stop() {
    stopExecutors()
    try {
      if (driverActor != null) {
        val future = driverActor.ask(StopDriver)(timeout)
        Await.ready(future, timeout)
      }
    } catch {
      case e: Exception =>
        throw new SparkException("Error stopping standalone scheduler's driver actor", e)
    }
  }

那么Master又是如何知道Driver Application退出的呢？这要归功于Akka的通讯机制了，当相互通讯的任意一方异常退出，另一方都会收到DisassociatedEvent, Master也就是在这个消息处理中移除已经停止的Driver Application。

    case DisassociatedEvent(_, address, _) => {
      // The disconnected client could've been either a worker or an app; remove whichever it was
      logInfo(s"$address got disassociated, removing it.")
      addressToWorker.get(address).foreach(removeWorker)
      addressToApp.get(address).foreach(finishApplication)
      if (state == RecoveryState.RECOVERING && canCompleteRecovery) { completeRecovery() }
    }

不告而别的方式下Executor是如何知道自己所服务的application已经顺利完成使命了呢？道理和master的一样，还是通过DisassociatedEvent来感知。详见CoarseGrainedExecutorBackend中的receive函数

  case x: DisassociatedEvent =>
      logError(s"Driver $x disassociated! Shutting down.")
      System.exit(1)

异常情况下的资源回收

由于Master和Worker之间的心跳机制，如果worker异常退出， Master会由心跳机制感知到其消亡，进而将其上报的资源移除。

Executor异常退出时，Worker中的监控线程ExecutorRunner会立即感知，进而上报给Master，Master会回收资源，并重新要求worker启动executor。