Spark 源码分析(四): Application 的注册

在前面一篇文章中分析到了 SparkContext 中的 TaskScheduler 创建及启动。

在 StandaloneSchedulerBackend start 代码里除了创建了一个 DriverEndpoint 用于 standalone 模式下用来和 Executor 通信之外还会创建一个 AppClient。

这个 AppClient 会向 Master 注册 Application,然后 Master 会通过 Application 的信息为它分配 Worker。

创建这个 AppClient 对象之前会去获取一些必要的参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
// 拿到 Driver RpcEndpoint 的地址
val driverUrl = RpcEndpointAddress(
sc.conf.get("spark.driver.host"),
sc.conf.get("spark.driver.port").toInt,
CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString
// 一些启动参数
val args = Seq(
"--driver-url", driverUrl,
"--executor-id", "{{EXECUTOR_ID}}",
"--hostname", "{{HOSTNAME}}",
"--cores", "{{CORES}}",
"--app-id", "{{APP_ID}}",
"--worker-url", "{{WORKER_URL}}")
// executor 额外的 java 选项
val extraJavaOpts = sc.conf.getOption("spark.executor.extraJavaOptions")
.map(Utils.splitCommandString).getOrElse(Seq.empty)
// executor 额外的环境变量
val classPathEntries = sc.conf.getOption("spark.executor.extraClassPath")
.map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)
// executor 额外的依赖
val libraryPathEntries = sc.conf.getOption("spark.executor.extraLibraryPath")
.map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)

val testingClassPath =
if (sys.props.contains("spark.testing")) {
sys.props("java.class.path").split(java.io.File.pathSeparator).toSeq
} else {
Nil
}

// Start executors with a few necessary configs for registering with the scheduler
val sparkJavaOpts = Utils.sparkJavaOpts(conf, SparkConf.isExecutorStartupConf)
val javaOpts = sparkJavaOpts ++ extraJavaOpts
// 将上面的那些信息封装成一个 command 对象
val command = Command("org.apache.spark.executor.CoarseGrainedExecutorBackend",
args, sc.executorEnvs, classPathEntries ++ testingClassPath, libraryPathEntries, javaOpts)
// 获取 application UI 的地址
val appUIAddress = sc.ui.map(_.appUIAddress).getOrElse("")
// 获取 executor 配置的 core 数量
val coresPerExecutor = conf.getOption("spark.executor.cores").map(_.toInt)

然后用上面的这些参数结合 SparkContext 中的一些数据封装一个 ApplicationDescription 对象,这对象里封装了一些信息,可以看看。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
private[spark] case class ApplicationDescription(
name: String,// 名字
maxCores: Option[Int],// 最多使用的 core 数量
memoryPerExecutorMB: Int,// 每个 Executor 分配的内存
command: Command,// 启动命令
appUiUrl: String,// application 的 UI Url 地址
eventLogDir: Option[URI] = None,// event 日志目录
// short name of compression codec used when writing event logs, if any (e.g. lzf)
eventLogCodec: Option[String] = None,
coresPerExecutor: Option[Int] = None,
// number of executors this application wants to start with,
// only used if dynamic allocation is enabled
initialExecutorLimit: Option[Int] = None,
user: String = System.getProperty("user.name", "<unknown>")) {

override def toString: String = "ApplicationDescription(" + name + ")"
}

封装好 ApplicationDescription 对象之后,根据这个对象创建一个 StandaloneAppClient 对象,然后调用 StandaloneAppClient 对象的 start 方法。

1
2
3
4
5
6
7
8
9
10
11
// 封装成一个 AppclientDescription 对象	
val appDesc = new ApplicationDescription(sc.appName, maxCores, sc.executorMemory, command,
appUIAddress, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor, initialExecutorLimit)
// 创建 StandaloneAppClient 对象
client = new StandaloneAppClient(sc.env.rpcEnv, masters, appDesc, this, conf)
// 调用 StandaloneAppClient 的 start 方法
client.start()
launcherBackend.setState(SparkAppHandle.State.SUBMITTED)
// 等待注册状态的更新
waitForRegistration()
launcherBackend.setState(SparkAppHandle.State.RUNNING)

StandaloneAppClient 的 start 方法会去创建并注册一个 ClientEndpoint 用来向 master 注册 Application。

1
2
3
4
def start() {
// Just launch an rpcEndpoint; it will call back into the listener.
endpoint.set(rpcEnv.setupEndpoint("AppClient", new ClientEndpoint(rpcEnv)))
}

ClientEndPoint 是一个 RpcEndpoint,在初始化的时候会去调用其 onstart 方法。

1
2
3
4
5
6
7
8
9
10
11
override def onStart(): Unit = {
try {
// 向 master 注册
registerWithMaster(1)
} catch {
case e: Exception =>
logWarning("Failed to connect to master", e)
markDisconnected()
stop()
}
}

registerWithMaster 方法实际上会去调用 tryRegisterAllMasters 方法,向所有的 Master 去注册。

在 Spark 中,Master 可能会是高可靠的 (HA),这种情况会有可能有多个 Master,不过只有一个 Master 处于 alive 状态,其它处于 standby 状态。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
/**
* 向所有的 master 进行一步注册,将会一直调用 tryRegisterAllMasters 方法进行注册,知道超出注册时间
* 当成功连接到 master ,所有调度的工作和 Futures 都会被取消
*/
private def registerWithMaster(nthRetry: Int) {
// 实际上调用了 tryRegisterAllMasters ,想所有 master 进行注册
registerMasterFutures.set(tryRegisterAllMasters())
registrationRetryTimer.set(registrationRetryThread.schedule(new Runnable {
override def run(): Unit = {
if (registered.get) {
registerMasterFutures.get.foreach(_.cancel(true))
registerMasterThreadPool.shutdownNow()
} else if (nthRetry >= REGISTRATION_RETRIES) {
markDead("All masters are unresponsive! Giving up.")
} else {
registerMasterFutures.get.foreach(_.cancel(true))
registerWithMaster(nthRetry + 1)
}
}
}, REGISTRATION_TIMEOUT_SECONDS, TimeUnit.SECONDS))
}

tryRegisterAllMasters 方法的调用,向所有 master 注册。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
// 异步向所有 master 注册,返回一个 [Future] 的数组用来以后取消
private def tryRegisterAllMasters(): Array[JFuture[_]] = {
for (masterAddress <- masterRpcAddresses) yield {
registerMasterThreadPool.submit(new Runnable {
override def run(): Unit = try {
if (registered.get) {
return
}
logInfo("Connecting to master " + masterAddress.toSparkURL + "...")
val masterRef = rpcEnv.setupEndpointRef(masterAddress, Master.ENDPOINT_NAME)
// 向 master 发送注册消息
masterRef.send(RegisterApplication(appDescription, self))
} catch {
case ie: InterruptedException => // Cancelled
case NonFatal(e) => logWarning(s"Failed to connect to master $masterAddress", e)
}
})
}
}

向 master 发送注册消息后,master 收到消息后注册完 application 之后会回复一条消息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
case RegisterApplication(description, driver) =>
// TODO Prevent repeated registrations from some driver
if (state == RecoveryState.STANDBY) {
// ignore, don't send response
} else {
logInfo("Registering app " + description.name)
val app = createApplication(description, driver)
// 将 master 中内存中等待调度的 app 队列更新,
registerApplication(app)
logInfo("Registered app " + description.name + " with ID " + app.id)
persistenceEngine.addApplication(app)
// 向 driver 回复一条注册 Application 的处理结果的消息
driver.send(RegisteredApplication(app.id, self))
// 调度资源
schedule()
}

master 调用 shedule 方法,这个方法做两件事,一个是给等待调度的 driver 分配资源,一个是给等待调度的 application 分配资源去启动 executor。

在给等待调度的 application 分配资源的时候最后会走到 launchExecutor 方法,这个方法会通过给符合要求的 worker 发送启动 executor 的消息去启动 executor。

1
2
3
4
5
6
7
8
9
private def launchExecutor(worker: WorkerInfo, exec: ExecutorDesc): Unit = {
logInfo("Launching executor " + exec.fullId + " on worker " + worker.id)
worker.addExecutor(exec)
// 给 worker 发送 LaunchExecutor 消息
worker.endpoint.send(LaunchExecutor(masterUrl,
exec.application.id, exec.id, exec.application.desc, exec.cores, exec.memory))
exec.application.driver.send(
ExecutorAdded(exec.id, worker.id, worker.hostPort, exec.cores, exec.memory))
}

在 worker 收到启动 Executor 的消息后会去根据消息去启动对应的 Executor。

至此,Application 的注册就完成了。