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

概要

本文主要讲述在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。