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