接着Spark 任务调度之Launch Driver，继续介绍Driver启动过程中，当SparkContext初始化时，Driver端注册DriverEndpoint到RpcEnv及Driver向Master注册APP信息的流程。
Register App
当 Driver 启动之后，就会运行 用户编写的代码的 Main方法，Main中首先 创建了 SparkContext，在 SparkContext 中 首先创建了 TaskScheduler，并调用 start 方法，如图：
在 Standalone 模式下，_taskScheduler 代表 StandaloneSchedulerBackend，_taskScheduler.start() 调用 StandaloneSchedulerBackend.scala 下的 start() 方法，先看下 StandaloneSchedulerBackend 类的定义：
private[spark] class StandaloneSchedulerBackend(
scheduler: TaskSchedulerImpl,
sc: SparkContext,
masters: Array[String])
extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv) {
...
}
可以看到 其继承自 CoarseGrainedSchedulerBackend。
StandaloneSchedulerBackend 的 start 方法：
// 步骤1
super.start()
...
// 步骤2 初始化 StandaloneAppClient，并调用 start 方法
client = new StandaloneAppClient(sc.env.rpcEnv, masters, appDesc, this, conf)
client.start()
首先 步骤1： 调用 父类也就是 CoarseGrainedSchedulerBackend（在下面子项中 介绍） 的 start 方法，主要是注册DriverEndpoint（在 CoarseGrainedSchedulerBackend.scala 下的一个类）到RpcEnv，DriverEndpoint用于提交task到Executor，接收Executor返回的计算结果。
其次 步骤2：注册ClientEndpoint，ClientEndpoint的生命周期方法onStart中会和 Master
通信，注册APP。
看下StandaloneAppClient 的 start() 方法：
def start() {
endpoint.set(rpcEnv.setupEndpoint("AppClient", new ClientEndpoint(rpcEnv)))
}
这样触发 ClientEndpoint 的 onStart 方法，onStart 中最终调用 tryRegisterAllMasters，主要是向 Master 发送注册App 的消息：
val masterRef = rpcEnv.setupEndpointRef(masterAddress, Master.ENDPOINT_NAME)
masterRef.send(RegisterApplication(appDescription, self))
看下 Master 接收 RegisterApplication 消息的处理：
//先创建Application，再注册
val app = createApplication(description, driver)
registerApplication(app)
看下 registerApplication：
apps += app
idToApp(app.id) = app
endpointToApp(app.driver) = app
addressToApp(appAddress) = app
waitingApps += app
完成流程如下（步骤3 SparkDeploySchedulerBackend 应该替换成 StandaloneSchedulerBackend，步骤6 AppClient 替换成 StandaloneAppClient）：
主要步骤如下
- 如上图①，Driver端注册DriverEndpoint到RpcEnv的流程，之后DriverEndpoint用于和Executor通信，包括send task和接收返回的计算结果。
- 如上图②，Driver向Master注册APP的流程。
下面介绍下 CoarseGrainedSchedulerBackend：
CoarseGrainedSchedulerBackend
CoarseGrainedSchedulerBackend 是粗粒度的SchedulerBackend实现，使用集合 executorDataMap 维护和 Executor 通信的RpcEndpointRef
executorDataMap
存储了 ExecutorData 的数据集
private val executorDataMap = new HashMap[String, ExecutorData]
看下ExecutorData ：
private[cluster] class ExecutorData(
val executorEndpoint: RpcEndpointRef,
val executorAddress: RpcAddress,
override val executorHost: String,
var freeCores: Int,
override val totalCores: Int,
override val logUrlMap: Map[String, String]
) extends ExecutorInfo(executorHost, totalCores, logUrlMap)
重点关注下面几个参数：
-
val executorEndpoint: RpcEndpointRef：负责发送消息给 ExecutorBackEnd
实际应用：
在 DriverEndpoint 的 receive方法中 KillTask 消息中 ，发送 killTask 消息：
executorDataMap.get(executorId) match {
executorInfo.executorEndpoint.send(KillTask(taskId, executorId, interruptThread, reason))
}
-
var freeCores: Int,：剩余核数
当Task 运行完，freeCores会增加：
if (TaskState.isFinished(state)) {
executorDataMap.get(executorId) match {
case Some(executorInfo) =>
executorInfo.freeCores += scheduler.CPUS_PER_TASK
}
}
当 Task 开始运行，freeCores会减少：
private def launchTasks(tasks: Seq[Seq[TaskDescription]]) {
for (task <- tasks.flatten) {
val executorData = executorDataMap(task.executorId)
executorData.freeCores -= scheduler.CPUS_PER_TASK
}
}
DriverEndpoint
DriverEndpoint 是 CoarseGrainedSchedulerBackend 的内部类
主要有两个职责：
职责一：
负责接收 var driverEndpoint: RpcEndpointRef = null 发送的消息。
主要在 CoarseGrainedSchedulerBackend 的 public 和
override 方法中调用 driverEndpoint send 方法：
def stopExecutors() {
driverEndpoint.askSync[Boolean](StopExecutors)
}
override def stop() {
driverEndpoint.askSync[Boolean](StopDriver)
}
driverEndpoint 在 CoarseGrainedSchedulerBackend 的 start 中创建，接着创建了 DriverEndpoint：
override def start() {
driverEndpoint = createDriverEndpointRef(properties)
}
protected def createDriverEndpointRef(
properties: ArrayBuffer[(String, String)]): RpcEndpointRef = {
rpcEnv.setupEndpoint(ENDPOINT_NAME, createDriverEndpoint(properties))
}
protected def createDriverEndpoint(properties: Seq[(String, String)]): DriverEndpoint = {
new DriverEndpoint(rpcEnv, properties)
}
DriverEndpoint：
class DriverEndpoint(override val rpcEnv: RpcEnv, sparkProperties: Seq[(String, String)])
extends ThreadSafeRpcEndpoint with Logging {
// 接收 RpcEndpointRef 发送的消息
override def receive: PartialFunction[Any, Unit] = {
case StatusUpdate =>
...
case ReviveOffers =>
...
case KillTask =>
...
}
}
职责二：
通过 ExecutorData 的 executorEndpoint 发送消息给 ExecutorBackEnd。主要包括：RegisteredExecutor、LaunchTask、KillTask、StopExecutor，一般和 DriverEndpoint 收到的事件对应。
