Yarn中如何实现ScheduleBackend

这篇文章将为大家详细讲解有关Yarn中如何实现ScheduleBackend，文章内容质量较高，因此小编分享给大家做个参考，希望大家阅读完这篇文章后对相关知识有一定的了解。

Yarn方式下的ScheduleBackend是用的啥？

在SparkContext中创建ScheduleBackend时，会根据指定的”master“参数的前缀决定创建哪种ScheduleBackend，对于"yarn://host:port"这样的URL来说，如果是cluster模式，就是创建YarnClusterSchedulerBackend，如果是client模式，就是创建YarnClientSchedulerBackend。

我们还是先看看YarnClusterSchedulerBackend的代码结构把。

YarnClusterSchedulerBackend继承了YarnSchedulerBackend，没有太多的发挥代码，我们直接看YarnSchedulerBackend把。估计client模式下也差不多。

YarnSchedulerBackend又继承了CoarseGrainedSchedulerBackend，我们看看不同点在哪里。

覆写了doRequestTotalExecutors和doKillExecutors方法，一个申请Executor，一个杀死Executor。

override def doRequestTotalExecutors(requestedTotal: Int): Future[Boolean] = {
    yarnSchedulerEndpointRef.ask[Boolean](prepareRequestExecutors(requestedTotal))
  }  
  override def doKillExecutors(executorIds: Seq[String]): Future[Boolean] = {
    yarnSchedulerEndpointRef.ask[Boolean](KillExecutors(executorIds))
  }

yarnSchedulerEndpointRef就是同一个文件里的endpoint端，看看具体的执行代码是什么：

      case r: RequestExecutors =>
        amEndpoint match {
          case Some(am) =>
            am.ask[Boolean](r).andThen {
              case Success(b) => context.reply(b)
              case Failure(NonFatal(e)) =>
                logError(s"Sending $r to AM was unsuccessful", e)
                context.sendFailure(e)
            }(ThreadUtils.sameThread)         
        }
      case k: KillExecutors =>
        amEndpoint match {
          case Some(am) =>
            am.ask[Boolean](k).andThen {
              case Success(b) => context.reply(b)
              case Failure(NonFatal(e)) =>
                logError(s"Sending $k to AM was unsuccessful", e)
                context.sendFailure(e)
            }(ThreadUtils.sameThread)          
        }

我们看到它又将消息转给了amEndpoint，就是转给了yarn工程里的ApplicationManager。又要跳到ApplicationManager去看看里面的实现逻辑了，真是一波三折啊。

ApplicationManager里是怎么处理RequestExecutors和KillExecutors两个消息的呢？

      case r: RequestExecutors =>
        Option(allocator) match {
          case Some(a) =>
            if (a.requestTotalExecutorsWithPreferredLocalities(r.requestedTotal,
              r.localityAwareTasks, r.hostToLocalTaskCount, r.nodeBlacklist)) {
              resetAllocatorInterval()
            }
            context.reply(true)
        }
      case KillExecutors(executorIds) =>
        Option(allocator) match {
          case Some(a) => executorIds.foreach(a.killExecutor)
        }
        context.reply(true)

调用allocator的killExecutor和requestTotalExecutorsWithPreferredLocalities方法。allocator又是啥？这里是不是类有的太多了啊。。

allocator = client.createAllocator(
      yarnConf,
      _sparkConf,
      appAttemptId,
      driverUrl,
      driverRef,
      securityMgr,
      localResources)

是client的createAllocator方法创建出来的，client是啥？是YarnRMClient，我们就要先看看YarnRMClient了，看名字就大概能猜到，YarnRMClient就是来向Yarn机器申请Executor和杀死Executor的。

createAllocator方法返回下面的YarnAllocator：

 return new YarnAllocator(driverUrl, driverRef, conf, sparkConf, amClient, appAttemptId, securityMgr,
      localResources, SparkRackResolver.get(conf))

来到YarnAllocator。

YarnAllocator的killExecutor方法很好理解，就是释放Yarn中的Container：

 def killExecutor(executorId: String): Unit = synchronized {
    executorIdToContainer.get(executorId) match {
      case Some(container) if !releasedContainers.contains(container.getId) =>
        internalReleaseContainer(container)
        runningExecutors.remove(executorId)
      case _ => logWarning(s"Attempted to kill unknown executor $executorId!")
    }
  }

申请Executor其实最终是在runAllocatedContainers方法中实现的。

核心代码看一下把，完整的可以看源码：

    if (runningExecutors.size() < targetNumExecutors) {
        numExecutorsStarting.incrementAndGet()
        if (launchContainers) {
          launcherPool.execute(() => {
            try {
              new ExecutorRunnable(
                Some(container),
                conf,
                sparkConf,
                driverUrl,
                executorId,
                executorHostname,
                executorMemory,
                executorCores,
                appAttemptId.getApplicationId.toString,
                securityMgr,
                localResources
              ).run()
              updateInternalState()
            } catch {              
            }
          })
        }

申请targetNumExecutors个ExecutorRunner，这样就和Standalone的申请Executor对应起来了。好了，整个过程就是这样了。

最终就会在Yarn集群中申请了所需数目的Container，并且在Container中启动ExecutorRunner，来向Driver汇报成绩。

这里的ExecutorRunner就是YarnCoarseGrainedExecutorBackend线程，在ExecutorRunner类中可以看到。

关于Yarn中如何实现ScheduleBackend就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。