Apache Spark源码走读之19 -- standalone cluster模式下资源的申请与释放
本帖最后由 pig2 于 2015-1-6 14:17 编辑问题导读
１．构成Standalone cluster部署模式的四大组成部件有哪些？分别有什么功能？
２．WorkerInfo在schedule函数中会被使用到，schedule函数处理逻辑是怎样的？
static/image/hrline/4.gif
概要本文主要讲述在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) {
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 / 1024 / 1024).toInt
} else {
val beanClass = Class.forName("com.sun.management.OperatingSystemMXBean")
val method = beanClass.getDeclaredMethod("getTotalPhysicalMemorySize")
totalMb = (method.invoke(bean).asInstanceOf / 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函数处理逻辑概述如下
[*]查看目前存活的worker中剩余的内存是否能够满足application每个task的最低需求，如果是则将该worker加入到可分配资源的队列
[*]根据分发策略，如果是决定将工作平摊到每个worker，则每次在一个worker上占用一个core，直到所有可分配资源耗尽或已经满足driver的需求
[*]如果分发策略是分发到尽可能少的worker，则一次占用尽worker上的可分配core，直到driver的core需求得到满足
[*]根据步骤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主要负责两件事情
[*]记录下新添加的executor使用掉的cpu core和内存数目，记录过程发生在worker.addExecutor
[*]向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。
相关内容
Apache Spark源码走读之1 -- Spark论文阅读笔记
Apache Spark源码走读之2 -- Job的提交与运行
Apache Spark源码走读之3-- Task运行期之函数调用关系分析
Apache Spark源码走读之4 -- DStream实时流数据处理
Apache Spark源码走读之5-- DStream处理的容错性分析
Apache Spark源码走读之6-- 存储子系统分析
Apache Spark源码走读之7 -- Standalone部署方式分析
Apache Spark源码走读之8 -- Spark on Yarn
Apache Spark源码走读之9 -- Spark源码编译
Apache Spark源码走读之10 -- 在YARN上运行SparkPi
Apache Spark源码走读之11 -- sql的解析与执行
Apache Spark源码走读之12 -- Hive on Spark运行环境搭建
Apache Spark源码走读之13 -- hiveql on spark实现详解
Apache Spark源码走读之14 -- Graphx实现剖析
Apache Spark源码走读之15 -- Standalone部署模式下的容错性分析
Apache Spark源码走读之16 -- spark repl实现详解
Apache Spark源码走读之17 -- 如何进行代码跟读
Apache Spark源码走读之18 -- 使用Intellij idea调试Spark源码
Apache Spark源码走读之20 -- ShuffleMapTask计算结果的保存与读取
Apache Spark源码走读之21 -- WEB UI和Metrics初始化及数据更新过程分析
Apache Spark源码走读之22 -- 浅谈mllib中线性回归的算法实现
Apache Spark源码走读之23 -- Spark MLLib中拟牛顿法L-BFGS的源码实现
Apache Spark源码走读之24 -- Sort-based Shuffle的设计与实现
本文转自徽沪一郎http://www.cnblogs.com/hseagle/p/3858694.html
{:soso_e179:} 学习。。 学习了，谢谢分享~
页:
[1]