Spark 之SparkContext 源码精读3

书接上文，SparkContext的CoarseGrainedSchedulerBackend已创建完毕，并且Driver也可以通过DriverEndpoint发消息了。

让咱们再回到CoarseGrainedSchedulerBackend的子类，SparkDeployScheduler.start方法中。至此super.start执行完毕。

继续往下看，获取driverUrl，也就是前述Rpc的地址；

第87行，创建了一个

1// SparkDeploySchedulerBackend.scala line 52
2override def start() {
3  super.start()
4  launcherBackend.connect()
5
6  // The endpoint for executors to talk to us
7  val driverUrl = rpcEnv.uriOf(SparkEnv.driverActorSystemName,
8    RpcAddress(sc.conf.get("spark.driver.host"), sc.conf.get("spark.driver.port").toInt),
9    CoarseGrainedSchedulerBackend.ENDPOINT_NAME)
10  val args = Seq(
11    "--driver-url", driverUrl,
12    "--executor-id", "{{EXECUTOR_ID}}",
13    "--hostname", "{{HOSTNAME}}",
14    "--cores", "{{CORES}}",
15    "--app-id", "{{APP_ID}}",
16    "--worker-url", "{{WORKER_URL}}")
17  val extraJavaOpts = sc.conf.getOption("spark.executor.extraJavaOptions")
18    .map(Utils.splitCommandString).getOrElse(Seq.empty)
19  val classPathEntries = sc.conf.getOption("spark.executor.extraClassPath")
20    .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)
21  val libraryPathEntries = sc.conf.getOption("spark.executor.extraLibraryPath")
22    .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)
23
24  // When testing, expose the parent class path to the child. This is processed by
25  // compute-classpath.{cmd,sh} and makes all needed jars available to child processes
26  // when the assembly is built with the "*-provided" profiles enabled.
27  val testingClassPath =
28    if (sys.props.contains("spark.testing")) {
29      sys.props("java.class.path").split(java.io.File.pathSeparator).toSeq
30    } else {
31      Nil
32    }
33
34  // Start executors with a few necessary configs for registering with the scheduler
35  val sparkJavaOpts = Utils.sparkJavaOpts(conf, SparkConf.isExecutorStartupConf)
36  val javaOpts = sparkJavaOpts ++ extraJavaOpts
37  // line 87
38  val command = Command("org.apache.spark.executor.CoarseGrainedExecutorBackend",
39    args, sc.executorEnvs, classPathEntries ++ testingClassPath, libraryPathEntries, javaOpts)
40  val appUIAddress = sc.ui.map(_.appUIAddress).getOrElse("")
41  val coresPerExecutor = conf.getOption("spark.executor.cores").map(_.toInt)
42  // line 91 
43  val appDesc = new ApplicationDescription(sc.appName, maxCores, sc.executorMemory,
44    command, appUIAddress, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor)
45  // line 93 
46  client = new AppClient(sc.env.rpcEnv, masters, appDesc, this, conf)
47  client.start()
48  launcherBackend.setState(SparkAppHandle.State.SUBMITTED)
49  waitForRegistration()
50  launcherBackend.setState(SparkAppHandle.State.RUNNING)
51}

然后就是一些赋值操作

在第81行，特别关键，创建了Command对象，进入Command定义，一看吓一跳啊。第一个参数名是mainClass，顾名思义，就是入口类。

1// org.apache.spark.deploy.Command line 22
2private[spark] case class Command(
3    mainClass: String,
4    arguments: Seq[String],
5    environment: Map[String, String],
6    classPathEntries: Seq[String],
7    libraryPathEntries: Seq[String],
8    javaOpts: Seq[String]) {
9}

先按住兴奋，回到SparkDeploySchedulerBackend.start继续往下看。至91行，将上面创建的Command及maxCores、sc.executorMemory,coresPerExecutors等一些参数，一起创建了ApplicationDescription。感觉里真相越来越近了。继续下去。

第93行，用上面创建的ApplicationDescription及master【Array[String]】等其他参数，一起创建了AppClient。注意，这个构造中，倒数第二个参数是将this传递进去了。

1// AppClient.scala line 33
2/**
3 * Interface allowing applications to speak with a Spark deploy cluster. Takes a master URL,
4 * an app description, and a listener for cluster events, and calls back the listener when various
5 * events occur.
6 *
7 * @param masterUrls Each url should look like spark://host:port.
8 */
9private[spark] class AppClient(
10    rpcEnv: RpcEnv,
11    masterUrls: Array[String],
12    appDescription: ApplicationDescription,
13    listener: AppClientListener,
14    conf: SparkConf)

注释很清晰的说明了，AppClient担任App和App所部署到的Spark集群沟通的角色。有master url ，app description，监听集群事件的监听器以及与各种监听到的集群的事件对应的回调。其中，倒数第二个参数是：AppClientListener，是一个trait。

1// AppClientListener.scala 
2/**
3 * Callbacks invoked by deploy client when various events happen. There are currently four events:
4 * connecting to the cluster, disconnecting, being given an executor, and having an executor
5 * removed (either due to failure or due to revocation).
6 * 
7 * Users of this API should *not* block inside the callback methods.
8 */
9private[spark] trait AppClientListener {
10  def connected(appId: String): Unit
11
12  /** Disconnection may be a temporary state, as we fail over to a new Master. */
13  def disconnected(): Unit
14
15  /** An application death is an unrecoverable failure condition. */
16  def dead(reason: String): Unit
17
18  def executorAdded(fullId: String, workerId: String, hostPort: String, cores: Int, memory: Int)
19
20  def executorRemoved(fullId: String, message: String, exitStatus: Option[Int]): Unit
21}

注释很清晰的说明了，处理4件事，连接上集群、与集群断开连接、新增一个Executor，一个Executor被移除。

而实际在代码中，还有一个dead回调。是对断开连接的补充，断开连接是临时状态，若故障转移成功，则又回到连接状态，若失败，则dead。

看到这，再看之前传入的参数，是SparkDeploySchedulerBackend将自实例传入，一定是他或他的父类实现了AppClientListener。

1// SparkDeploySchedulerBackend.scala 
2private[spark] class SparkDeploySchedulerBackend(
3    scheduler: TaskSchedulerImpl,
4    sc: SparkContext,
5    masters: Array[String])
6  extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv)
7  with AppClientListener
8  with Logging

果然如此，实现了AppClientListener。

思路回到创建SparkDeploySchedulerBackend的 第94行。执行了AppClient.start

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

又创建了ClientEndpoint。

1// AppClient.scala line 57
2private class ClientEndpoint(override val rpcEnv: RpcEnv) extends ThreadSafeRpcEndpoint
3  with Logging

请回忆前面的内容，是否有点眼熟？没错，Endpoint，实现了ThreadSafeRpcEndpoint。有印象吗？没错，是DriverEndpoint。

同样的RpcEndpoint的生命周期是什么？复习下，没错，构造-> onStart ->receive* ->onStop。

了解下onStart，顾名思义，registerWithMaster，向Master注册。进入registerWithMaster瞅瞅。

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

进入registerWithMaster瞅瞅。

结合下面的代码和注释，很清晰的知道，Driver以多线程异步的方式向Master注册，一旦任意一个注册成功，则其他注册取消。若注册超时 ${REGISTRATION_TIMEOUT_SECONDS}，会重试${REGISTRATION_RETRIES}次。

1// AppClient.scala line 125
2/**
3 * Register with all masters asynchronously. It will call `registerWithMaster` every
4 * REGISTRATION_TIMEOUT_SECONDS seconds until exceeding REGISTRATION_RETRIES times.
5 * Once we connect to a master successfully, all scheduling work and Futures will be cancelled.
6 *
7 * 
8 */
9private def registerWithMaster(nthRetry: Int) {
10  registerMasterFutures.set(tryRegisterAllMasters())
11  registrationRetryTimer.set(registrationRetryThread.scheduleAtFixedRate(new Runnable {
12    override def run(): Unit = {
13      Utils.tryOrExit {
14        if (registered.get) {
15          registerMasterFutures.get.foreach(_.cancel(true))
16          registerMasterThreadPool.shutdownNow()
17        } else if (nthRetry >= REGISTRATION_RETRIES) {
18          markDead("All masters are unresponsive! Giving up.")
19        } else {
20          registerMasterFutures.get.foreach(_.cancel(true))
21          registerWithMaster(nthRetry + 1)
22        }
23      }
24    }
25  }, REGISTRATION_TIMEOUT_SECONDS, REGISTRATION_TIMEOUT_SECONDS, TimeUnit.SECONDS))
26}
27
28// AppClient.scala line 96
29/**
30 *  Register with all masters asynchronously and returns an array `Future`s for cancellation.
31 */
32private def tryRegisterAllMasters(): Array[JFuture[_]] = {
33  for (masterAddress <- masterRpcAddresses) yield {
34    registerMasterThreadPool.submit(new Runnable {
35      override def run(): Unit = try {
36        if (registered.get) {
37          return
38        }
39        logInfo("Connecting to master " + masterAddress.toSparkURL + "...")
40        val masterRef =
41          rpcEnv.setupEndpointRef(Master.SYSTEM_NAME, masterAddress, Master.ENDPOINT_NAME)
42// line 109
43        masterRef.send(RegisterApplication(appDescription, self))
44      } catch {
45        case ie: InterruptedException => // Cancelled
46        case NonFatal(e) => logWarning(s"Failed to connect to master $masterAddress", e)
47      }
48    })
49  }
50}

具体看下消息的发收，AppClient的109 行，向master发送了一个RegisterApplication类型的消息。

1// DeployMessages.scala line 108
2case class RegisterApplication(appDescription: ApplicationDescription, driver: RpcEndpointRef)
3  extends DeployMessage

接收者是Master，回忆前面的RpcEndpoint，接收消息是receive*

1// Master.scala line 244
2case RegisterApplication(description, driver) => {
3  // TODO Prevent repeated registrations from some driver
4  if (state == RecoveryState.STANDBY) {
5    // ignore, don't send response
6  } else {
7    logInfo("Registering app " + description.name)
8    val app = createApplication(description, driver)
9// line 251
10    registerApplication(app)
11    logInfo("Registered app " + description.name + " with ID " + app.id)
12// line 253
13    persistenceEngine.addApplication(app)
14    driver.send(RegisteredApplication(app.id, self))
15    schedule()
16  }
17}

将Application注册->持久化信息以备恢复->发一个RegisteredApplication消息给Driver->资源调度。

1// Master.scala line 795
2private def registerApplication(app: ApplicationInfo): Unit = {
3  val appAddress = app.driver.address
4  if (addressToApp.contains(appAddress)) {
5    logInfo("Attempted to re-register application at same address: " + appAddress)
6    return
7  }
8  applicationMetricsSystem.registerSource(app.appSource)
9  apps += app
10  idToApp(app.id) = app
11  endpointToApp(app.driver) = app
12  addressToApp(appAddress) = app
13  waitingApps += app
14}

简单的数据设置。master持久化driver后续会细细分析。

持久化成功后，Master会发送一条RegisteredApplication类型的消息给Driver，理解为你让我办的事，妥了。

1// DeployMessages.scala line 121
2case class RegisteredApplication(appId: String, master: RpcEndpointRef) extends DeployMessage

Driver接收到消息

1// AppClient.scala line 160
2case RegisteredApplication(appId_, masterRef) =>
3  appId.set(appId_)
4  registered.set(true)
5  master = Some(masterRef)
6  listener.connected(appId.get)

Driver收到消息后，简单的做了些设置。理解为 "哦"。

至此，Driver向Master注册已经成功。

其实，本篇博客另一个关键的信息在于，展示了部署类消息都在 org.apache.spark.deploy.DeployMessages.scala。可以通过不同类型的消息，寻找对应的发送者和接收者，将流程贯穿起来。

如有理解错误，请大家回复我修改。谢谢。