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就分享到这里了，希望以上内容可以对大家有一定的帮助，可以学到更多知识。如果觉得文章不错，可以把它分享出去让更多的人看到。