Spark 源码 | Standalone Client 模式提交流程（七）

前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住给大家分享一下。点击跳转到网站：https://www.captainai.net/dongkelun

欢 迎 关 注 我 的 公 众 号

前言

在第五篇文章 Yarn Client 模式提交流程（五） 中，我们分析了 Yarn Client 模式的完整提交流程。本文将继续分析 Spark 的另一种常用部署模式——Standalone Client 模式的提交流程。

Standalone 模式是 Spark 自带的集群管理器，不依赖 YARN、Mesos 等外部资源调度系统。其架构包含 Master（集群管理器）和 Worker（工作节点）两个角色，Master 负责资源调度，Worker 负责执行计算任务。

关于 Standalone 集群的搭建与配置，可参考之前的文章 Spark Standalone 集群配置。Master 和 Worker 的启动源码分析将在后续文章中详细介绍。

版本

Spark 3.2.3

Standalone Client 模式概述

Standalone Client 模式下，Driver 运行在提交应用的客户端机器上，向 Master 注册 Application，Master 调度 Worker 启动 Executor。

与 Yarn Client 的直观对比：

Yarn Client 模式中的 ResourceManager + NodeManager 对应到 Standalone 模式中的 Master + Worker。区别在于 Standalone 模式是 Spark 自带的简易调度系统，功能比 YARN 精简，但启动和调试更加方便。

模式	Master 地址	childMainClass	Driver 位置
Standalone Client	spark://host:port	args.mainClass（用户主类）	客户端
Standalone Cluster	spark://host:port	SparkApplication 或 Wrapper	集群
Yarn Client	yarn	args.mainClass（用户主类）	客户端

// 提交命令示例
spark-submit --master spark://host:7077 --deploy-mode client \
  --class org.apache.spark.examples.SparkPi \
  $SPARK_HOME/examples/jars/spark-examples_2.12-3.2.3.jar

0. RPC 通信方式汇总

本文多处涉及 Spark RPC 通信，这里做一个统一说明。完整的 Spark RPC 框架总结见 Spark RPC 学习总结：

• rpcEnv.setupEndpoint(name, endpoint)：注册本地 RpcEndpoint，注册完成后自动触发 endpoint.onStart()。之后可通过返回的 RpcEndpointRef 收发消息。
• rpcEnv.setupEndpointRef(address, name)：获取远程 RpcEndpoint 的引用，用于向远程端点发送消息。
• ref.send(msg)：异步发送消息，无返回值。对应 RpcEndpoint.receive。用于”通知”类消息。
• ref.ask[T](msg) / ref.askSync[T](msg)：发送消息并等待返回值。对应 RpcEndpoint.receiveAndReply。用于”请求-响应”类消息。

本文涉及的 RPC 通信对照：

位置	方式	原因
5.2 Client → Master 注册：masterRef.send(RegisterApplication)	send	只需通知，不阻塞等待，Master 成功后通过回调回复
6.1 Master → Client 回复：driver.send(RegisteredApplication)	send	同上，单向通知
6.5 Master → Worker：worker.endpoint.send(LaunchExecutor)	send	通知 Worker 启动 Executor
6.5 Worker → Driver：exec.application.driver.send(ExecutorAdded)	send	通知 Driver 有新的 Executor 被分配
8.2 Executor 获取配置：driverRef.askSync(RetrieveSparkAppConfig)	askSync	需要获取配置内容，必须等结果
8.2 Executor 反向注册：ref.ask[Boolean](RegisterExecutor(...))	ask	需要知道 Driver 是否确认注册成功

本文涉及的 RPC Endpoint 名称及注册位置：

角色	RPC Endpoint 名称	注册代码	对应类
Master	"Master"	Master.scala setupEndpoint("Master", ...)	Master
Worker	"Worker"	Worker.scala setupEndpoint("Worker", ...)	Worker
Driver 端	"CoarseGrainedScheduler"（即 CoarseGrainedSchedulerBackend.ENDPOINT_NAME）	CoarseGrainedSchedulerBackend.scala setupEndpoint(ENDPOINT_NAME, ...)	DriverEndpoint
Executor 端	"Executor"	CoarseGrainedExecutorBackend.scala setupEndpoint("Executor", backend)	CoarseGrainedExecutorBackend
AppClient	"AppClient"	StandaloneAppClient.scala setupEndpoint("AppClient", ...)	ClientEndpoint

• --driver-url spark://CoarseGrainedScheduler@driverHost:driverPort 中的 CoarseGrainedScheduler 就是这个 RPC Endpoint 名称，不是类名
• Executor 启动后以 "Executor" 名称注册 RPC Endpoint，与 "CoarseGrainedScheduler" 对称

完整提交流程

1. SparkSubmit 准备阶段

在第二篇文章 SparkSubmit 提交流程分析（二） 中，我们分析了 prepareSubmitEnvironment 方法。该方法根据 --master 和 --deploy-mode 设置 childMainClass，决定 main 方法在哪里执行。

对于 Standalone Client 模式，关键代码在 SparkSubmit.scala 的 prepareSubmitEnvironment 方法中。当 deployMode == CLIENT 时，设置 childMainClass：

// SparkSubmit.scala — prepareSubmitEnvironment 方法

// In client mode, launch the application main class directly
// 在客户端模式下，childMainClass 直接等于用户传入的主类
// In addition, add the main application jar and any added jars (if any) to the classpath
// 此外，将主应用程序jar和任何添加的jar（如果有的话）添加到类路径中
if (deployMode == CLIENT) {
  childMainClass = args.mainClass           // ← 核心：直接使用用户主类
  if (localPrimaryResource != null && isUserJar(localPrimaryResource)) {
    childClasspath += localPrimaryResource
  }
  if (localJars != null) { childClasspath ++= localJars.split(",") }
}

关键点：Standalone Client 和 Yarn Client 在这步完全一致——childMainClass 都设为 args.mainClass，也都是在客户端通过 runMain 反射执行用户主类的 main 方法。Driver 在提交客户端运行。

整理后有以下对应关系：

模式	childMainClass	main 方法执行位置	Driver 位置
Standalone Client	args.mainClass（用户主类）	runMain 在客户端反射调用	客户端
Standalone Cluster	SparkApplication 或 Wrapper	runDriver 在 Worker 内调用	集群

2. runMain 执行用户主类

prepareSubmitEnvironment 返回后，runMain 方法通过反射加载并执行用户主类的 main 方法：

// SparkSubmit.scala — runMain 方法
private def runMain(
    args: SparkSubmitArguments,
    uninitLog: Boolean): Unit = {

  // ① 获取 prepareSubmitEnvironment 的返回值
  val (childArgs, childClasspath, sparkConf, childMainClass) = prepareSubmitEnvironment(args)

  // ② 通过反射加载主类
  var mainClass: Class[_] = null
  mainClass = Utils.classForName(childMainClass)

  // ③ 包装主类（如果实现了 SparkApplication 则直接使用，否则包装成 JavaMainApplication）
  val app: SparkApplication = if (classOf[SparkApplication].isAssignableFrom(mainClass)) {
    mainClass.getConstructor().newInstance().asInstanceOf[SparkApplication]
  } else {
    new JavaMainApplication(mainClass)
  }

  // ④ 调用 app.start 启动应用
  app.start(childArgs.toArray, sparkConf)
}

JavaMainApplication.start 内部通过反射调用用户主类的 main 方法：

// SparkSubmit.scala
class JavaMainApplication(mainClass: Class[_]) extends SparkApplication {
  override def start(args: Array[String], conf: SparkConf): Unit = {
    // 通过反射调用用户主类的 main 方法，Driver 就此启动
    val mainMethod = mainClass.getMethod("main", classOf[Array[String]])
    mainMethod.invoke(null, args)
  }
}

用户主类 main 方法中的典型调用链：

用户主类.main
  → SparkSession.builder().getOrCreate()
    → 内部调用 SparkContext 的构造方法

3. SparkContext 初始化与 TaskScheduler 创建

SparkContext 初始化时，会创建 TaskScheduler 和 SchedulerBackend。在第四篇文章 SparkContext 初始化与 TaskScheduler 创建流程（四） 中已经详细分析了该流程。

当 master = "spark://..." 时，通过 createTaskScheduler 方法创建的是：

• TaskScheduler：TaskSchedulerImpl
• SchedulerBackend：StandaloneSchedulerBackend

createTaskScheduler 方法

createTaskScheduler 是 SparkContext 内部方法，根据 master URL 创建对应的 TaskScheduler 和 SchedulerBackend：

// SparkContext.scala
private def createTaskScheduler(
    sc: SparkContext,
    master: String,
    deployMode: String): (SchedulerBackend, TaskScheduler) = {
  master match {
    case "local" =>
      val scheduler = new TaskSchedulerImpl(sc, MAX_LOCAL_TASK_FAILURES, isLocal = true)
      val backend = new LocalSchedulerBackend(sc.getConf, scheduler, 1)
      scheduler.initialize(backend)
      (backend, scheduler)

    case SPARK_REGEX(sparkUrl) =>
      // Standalone 模式 - 直接创建，不经过 SPI
      val scheduler = new TaskSchedulerImpl(sc)
      val masterUrls = sparkUrl.split(",").map("spark://" + _)
      val backend = new StandaloneSchedulerBackend(scheduler, sc, masterUrls)
      scheduler.initialize(backend)
      (backend, scheduler)

    case masterUrl =>
      // 其他模式（YARN、Kubernetes、Mesos）：通过 ExternalClusterManager SPI 加载
      val cm = getClusterManager(masterUrl) match {
        case Some(clusterMgr) => clusterMgr
        case None => throw new SparkException("Could not parse Master URL: '" + master + "'")
      }
      val scheduler = cm.createTaskScheduler(sc, masterUrl)
      val backend = cm.createSchedulerBackend(sc, masterUrl, scheduler)
      cm.initialize(scheduler, backend)
      (backend, scheduler)
  }
}

关键点：Standalone 模式走的是 SPARK_REGEX 分支，直接在 createTaskScheduler 内部创建 TaskSchedulerImpl + StandaloneSchedulerBackend。而 YARN 模式走的是 masterUrl 兜底分支，通过 Java SPI 机制加载 ExternalClusterManager 实现类来创建，具体分析见第四篇文章。

创建完成后，SparkContext 初始化过程中会调用 _taskScheduler.start() → TaskSchedulerImpl.start() 内部调用 backend.start() 即 StandaloneSchedulerBackend.start()，正式进入提交流程的核心。

4. StandaloneSchedulerBackend.start() — 核心入口

StandaloneSchedulerBackend 继承自 CoarseGrainedSchedulerBackend，实现了 StandaloneAppClientListener 接口。它的 start() 方法是整个 Standalone Client 提交流程的核心入口：

// StandaloneSchedulerBackend.scala
private[spark] class StandaloneSchedulerBackend(
    scheduler: TaskSchedulerImpl,
    sc: SparkContext,
    masters: Array[String])
  extends CoarseGrainedSchedulerBackend(scheduler, sc.env.rpcEnv)
  with StandaloneAppClientListener
  with Logging {

  private[spark] var client: StandaloneAppClient = null
  private val registrationBarrier = new Semaphore(0)  // 注册阻塞信号量

  override def start(): Unit = {
    // ① 调用父类 CoarseGrainedSchedulerBackend.start()
    super.start()

    // ② LauncherBackend：Client 模式才连接
    if (sc.deployMode == "client") {
      launcherBackend.connect()
    }

    // ③ 构建 Driver URL，Executor 通过此 URL 连回 DriverEndpoint
    val driverUrl = RpcEndpointAddress(
      sc.conf.get(config.DRIVER_HOST_ADDRESS),
      sc.conf.get(config.DRIVER_PORT),
      CoarseGrainedSchedulerBackend.ENDPOINT_NAME).toString

    // ④ 构建 Executor 启动命令模板
    //    使用 {{EXECUTOR_ID}}、{{HOSTNAME}} 等占位符
    //    Worker 在真正启动 Executor 时会替换这些占位符
    val args = Seq(
      "--driver-url", driverUrl,
      "--executor-id", "{{EXECUTOR_ID}}",
      "--hostname", "{{HOSTNAME}}",
      "--cores", "{{CORES}}",
      "--app-id", "{{APP_ID}}",
      "--worker-url", "{{WORKER_URL}}")
    val extraJavaOpts = sc.conf.get(config.EXECUTOR_JAVA_OPTIONS)
      .map(Utils.splitCommandString).getOrElse(Seq.empty)
    val classPathEntries = sc.conf.get(config.EXECUTOR_CLASS_PATH)
      .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)
    val libraryPathEntries = sc.conf.get(config.EXECUTOR_LIBRARY_PATH)
      .map(_.split(java.io.File.pathSeparator).toSeq).getOrElse(Nil)

    // ⑤ 构建 Executor 启动命令
    //    启动入口：CoarseGrainedExecutorBackend（与 YARN 不同，YARN 用的是 YarnCoarseGrainedExecutorBackend）
    val command = Command(
      "org.apache.spark.executor.CoarseGrainedExecutorBackend",
      args, sc.executorEnvs, classPathEntries ++ testingClassPath,
      libraryPathEntries, javaOpts)

    val webUrl = sc.ui.map(_.webUrl).getOrElse("")
    val coresPerExecutor = conf.getOption(config.EXECUTOR_CORES.key).map(_.toInt)
    val initialExecutorLimit =
      if (Utils.isDynamicAllocationEnabled(conf)) Some(0) else None
    val executorResourceReqs = ResourceUtils.parseResourceRequirements(conf,
      config.SPARK_EXECUTOR_PREFIX)

    // ⑥ 创建 ApplicationDescription
    val appDesc = ApplicationDescription(sc.appName, maxCores, sc.executorMemory, command,
      webUrl, sc.eventLogDir, sc.eventLogCodec, coresPerExecutor, initialExecutorLimit,
      resourceReqsPerExecutor = executorResourceReqs)

    // ⑦ 创建 StandaloneAppClient——核心客户端，负责与 Master 通信
    client = new StandaloneAppClient(sc.env.rpcEnv, masters, appDesc, this, conf)

    // ⑧ 启动 Client
    client.start()

    launcherBackend.setState(SparkAppHandle.State.SUBMITTED)

    // ⑨ 阻塞等待注册成功（信号量）
    waitForRegistration()

    launcherBackend.setState(SparkAppHandle.State.RUNNING)
  }

4.1 启动命令模板与 CoarseGrainedExecutorBackend

从代码中可以看到，Standalone 模式下的 Executor 启动入口类为：

org.apache.spark.executor.CoarseGrainedExecutorBackend

与 YARN 模式有所不同：

模式	Executor 入口类
Standalone	CoarseGrainedExecutorBackend
Yarn Client	YarnCoarseGrainedExecutorBackend（继承自 CoarseGrainedExecutorBackend）

CoarseGrainedExecutorBackend 的启动命令分两阶段构建：

阶段一（Driver 端）：构建命令模板 — StandaloneSchedulerBackend.start()

// StandaloneSchedulerBackend.scala — start() 方法

// ④ 构建 Executor 启动命令模板
// driverUrl 此时已经是真实值（driverHost + driverPort 已确定）
// 其余参数用 {{}} 占位符，Worker 端替换
val args = Seq(
  "--driver-url", driverUrl,          // 实际地址，如 spark://CoarseGrainedScheduler@host:port
  "--executor-id", "{{EXECUTOR_ID}}", // 占位符
  "--hostname", "{{HOSTNAME}}",       // 占位符
  "--cores", "{{CORES}}",             // 占位符
  "--app-id", "{{APP_ID}}",           // 占位符
  "--worker-url", "{{WORKER_URL}}")   // 占位符

// 封装为 Command 对象，再封装到 ApplicationDescription 中
val command = Command(
  "org.apache.spark.executor.CoarseGrainedExecutorBackend",  // 主类
  args,    // ← 上面的参数列表
  ...      // extraJavaOpts, classPathEntries, libraryPathEntries
)

val appDesc = ApplicationDescription(sc.appName, maxCores, sc.executorMemory, command, ...)

这个 ApplicationDescription 通过 RegisterApplication 消息发送给 Master，Master 调度时传给 Worker。

阶段二（Worker 端）：替换占位符拼出最终命令 — ExecutorRunner.fetchAndRunExecutor()

// ExecutorRunner.scala — fetchAndRunExecutor() 方法

// 从 ApplicationDescription 中取出命令参数
val arguments = appDesc.command.arguments

// 替换 javaOpts 中的 {{APP_ID}} 和 {{EXECUTOR_ID}}
val subsOpts = appDesc.command.javaOpts.map {
  Utils.substituteAppNExecIds(_, appId, execId.toString)
}

// 用替换后的参数重建 Command
val subsCommand = appDesc.command.copy(arguments = arguments, javaOpts = subsOpts)

// 构建实际进程（内部调用 substituteVariables 替换占位符）
val builder = CommandUtils.buildProcessBuilder(subsCommand, ...)

其中占位符替换的核心方法：

// ExecutorRunner.scala — 占位符替换
private[worker] def substituteVariables(argument: String): String = argument match {
  case "{{WORKER_URL}}" => workerUrl   // workerUrl 由 ExecutorRunner 构造函数传入
  case "{{EXECUTOR_ID}}" => execId.toString
  case "{{HOSTNAME}}" => host
  case "{{CORES}}" => cores.toString
  case "{{APP_ID}}" => appId
  case other => other
}

为什么 --driver-url 不是占位符？因为 Driver 地址在 StandaloneSchedulerBackend.start() 时已经确定（driverHost + driverPort），不需要等到 Worker 端再填。而 executorId、hostname 等要等到 Worker 真正分配资源时才知道实际值。

替换后的最终命令见第 7.2 节。

4.2 ApplicationDescription

ApplicationDescription 封装了应用的描述信息，包括应用名称、需要的 CPU/内存、Executor 启动命令等，Master 将根据这些信息进行资源调度：

val appDesc = ApplicationDescription(
  sc.appName,           // 应用名称
  maxCores,             // 最大核心数
  sc.executorMemory,    // Executor 内存
  command,              // Executor 启动命令（CoarseGrainedExecutorBackend）
  webUrl,               // Web UI 地址
  sc.eventLogDir,       // 事件日志目录
  sc.eventLogCodec,     // 事件日志压缩编码
  coresPerExecutor,     // 每个 Executor 的核心数
  initialExecutorLimit, // 初始 Executor 限制（动态分配时为 0）
  resourceReqsPerExecutor = executorResourceReqs  // 资源需求
)

5. StandaloneAppClient 注册流程

RPC 基础知识（setupEndpoint / send / ask 等）见第 0 章「RPC 通信方式汇总」。

StandaloneAppClient 是 Driver 端与 Master 通信的客户端。创建后，调用 client.start() 触发注册流程：

// StandaloneAppClient.scala
class StandaloneAppClient(
    rpcEnv: RpcEnv,
    masterUrls: Array[String],
    appDescription: ApplicationDescription,
    listener: StandaloneAppClientListener,  // ← 这里传入的是 StandaloneSchedulerBackend 自身
    conf: SparkConf) extends Logging {

  def start(): Unit = {
    // 创建 ClientEndpoint（RPC 端点）并注册到 RpcEnv
    endpoint.set(rpcEnv.setupEndpoint("AppClient", new ClientEndpoint(rpcEnv)))
  }
}

5.1 ClientEndpoint.onStart()

setupEndpoint 注册 RpcEndpoint 时自动触发 onStart()（见 0 RPC 基础），开始注册流程：

// StandaloneAppClient.scala — ClientEndpoint 内部类

private val REGISTRATION_TIMEOUT_SECONDS = 20  // 超时时间 20 秒
private val REGISTRATION_RETRIES = 3           // 最大重试次数 3 次

override def onStart(): Unit = {
  try {
    // 开始注册，传入重试次数 1（首次尝试）
    registerWithMaster(1)
  } catch {
    case e: Exception =>
      logWarning("Failed to connect to master", e)
      markDisconnected()
      stop()
  }
}

5.2 registerWithMaster() 与 tryRegisterAllMasters()

registerWithMaster() 调用 tryRegisterAllMasters() 向所有 Master 地址发送注册请求，并设置定时器处理重试：

// StandaloneAppClient.scala — ClientEndpoint

/**
 * 向所有 Master 异步注册。如果注册失败，每 REGISTRATION_TIMEOUT_SECONDS 秒
 * 重试一次，最多 REGISTRATION_RETRIES 次。
 *
 * @param nthRetry 第几次重试（从 1 开始）
 */
private def registerWithMaster(nthRetry: Int): Unit = {
  // ① 向所有 Master 发送注册请求（并行）
  registerMasterFutures.set(tryRegisterAllMasters())

  // ② 设置定时器：20 秒后检查注册状态
  registrationRetryTimer.set(registrationRetryThread.schedule(new Runnable {
    override def run(): Unit = {
      if (registered.get) {
        // 已注册成功 → 取消所有未完成的 Future
        registerMasterFutures.get.foreach(_.cancel(true))
        registerMasterThreadPool.shutdownNow()
      } else if (nthRetry >= REGISTRATION_RETRIES) {
        // 重试超过 3 次 → 宣告失败
        markDead("All masters are unresponsive! Giving up.")
      } else {
        // 取消当前请求，继续重试
        registerMasterFutures.get.foreach(_.cancel(true))
        registerWithMaster(nthRetry + 1)
      }
    }
  }, REGISTRATION_TIMEOUT_SECONDS, TimeUnit.SECONDS))
}

重试机制总结：

• 最大重试次数：3 次
• 重试间隔：20 秒
• 总超时：最多 60 秒（3 × 20s）
• 并发：同时向所有 Master 地址发起注册

为什么要向多个 Master 注册？

核心原因：Spark Standalone Master HA（高可用）机制

虽然单机测试时只有一个 Master，但生产环境中 Master 是单点故障（SPOF）。Spark 通过以下方式支持 Master HA：

1. 配置多个 Master 地址：spark.master 参数支持逗号分隔多个地址，如：

   spark.master=spark://master1:7077,master2:7077,master3:7077

   典型 HA 部署为一主一备两个 Master，但数组不做数量限制，理论上可多于两个（实际极少需要）
2. ZooKeeper 协调：HA 模式下多个 Master 节点通过 ZooKeeper 选举一个 Active Master，其余为 Standby
3. masterUrls 的解析：

   // 支持逗号分隔的多个 Master 地址，用于 Standalone Master HA 场景，典型一主一备
   val masterUrls = sparkUrl.split(",").map("spark://" + _)

   每个地址都会作为 StandaloneAppClient.masterRpcAddresses 中的一个元素
4. “广撒网”策略：
   • Client/Worker 启动时不知道哪个 Master 是 Active
   • 并行向所有配置的 Master 发送注册请求
   • 只有 Active Master 会响应 RegisteredApplication
   • Standby Master 会忽略或拒绝注册
   • 注册成功后，后续通信只与 Active Master 进行

Worker 端也采用完全相同的策略向所有 Master 注册，原理一致。

tryRegisterAllMasters() 向所有 Master 地址发送 RegisterApplication 消息：

// StandaloneAppClient.scala — ClientEndpoint

/**
 * 向所有 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 + "...")
        // 获取 Master 的 RpcEndpointRef
        val masterRef = rpcEnv.setupEndpointRef(masterAddress, Master.ENDPOINT_NAME)
        // 发送 RegisterApplication 消息
        masterRef.send(RegisterApplication(appDescription, self))
      } catch {
        case ie: InterruptedException => // 取消
        case NonFatal(e) => logWarning(s"Failed to connect to master $masterAddress", e)
      }
    })
  }
}

5.3 注册成功回调

当 Master 回复 RegisteredApplication 消息后，ClientEndpoint.receive 处理该消息：

// StandaloneAppClient.scala — ClientEndpoint.receive

override def receive: PartialFunction[Any, Unit] = {
  case RegisteredApplication(appId_, masterRef) =>
    // 保存 appId
    appId.set(appId_)
    // 标记已注册
    registered.set(true)
    // 记录当前 Master 引用
    master = Some(masterRef)
    // 回调 listener.connected()，释放信号量
    listener.connected(appId.get)

  // ... 其他消息
}

listener.connected() 对应 StandaloneSchedulerBackend.connected()：

// StandaloneSchedulerBackend.scala

override def connected(appId: String): Unit = {
  logInfo("Connected to Spark cluster with app ID " + appId)
  this.appId = appId
  notifyContext()          // release() → 释放信号量，waitForRegistration()的acquire()返回
  launcherBackend.setAppId(appId)
}

// 同一信号量的两端：waitForRegistration 阻塞，notifyContext 释放
private def waitForRegistration() = {
  registrationBarrier.acquire()  // 阻塞等待信号量
}

private def notifyContext() = {
  registrationBarrier.release()  // 释放信号量 → acquire 返回
}

至此，StandaloneSchedulerBackend.start() 中的 waitForRegistration() 返回，注册完成。Driver 端继续执行后续逻辑。

注册流程时序：

StandaloneSchedulerBackend.start()
  → client.start() → setupEndpoint("AppClient", ClientEndpoint)
    → ClientEndpoint.onStart()
      → registerWithMaster(1)
        → tryRegisterAllMasters()
          → masterRef.send(RegisterApplication(appDesc, self))

[Master 处理注册，回复 RegisteredApplication]

        → receive: RegisteredApplication
          → listener.connected(appId)
            → registrationBarrier.release()

waitForRegistration() 返回 → 注册完成

6. Master 注册与调度

6.1 Master.receive — RegisterApplication

Master 收到 RegisterApplication 消息后，处理流程如下：

// Master.scala — receive 方法

case RegisterApplication(description, driver) =>
  if (state == RecoveryState.STANDBY) {
    // Standby 状态忽略注册请求
  } else {
    logInfo("Registering app " + description.name)

    // ① 创建 ApplicationInfo（分配 appId，如 app-20260718-0000）
    val app = createApplication(description, driver)

    // ② 注册 Application（加入 apps 和 waitingApps 集合）
    registerApplication(app)

    logInfo("Registered app " + description.name + " with ID " + app.id)

    // ③ 持久化 Application（用于故障恢复）
    persistenceEngine.addApplication(app)

    // ④ 回复注册成功
    driver.send(RegisteredApplication(app.id, self))

    // ⑤ 触发资源调度
    schedule()
  }

createApplication 方法创建 ApplicationInfo：

// Master.scala
private def createApplication(desc: ApplicationDescription, driver: RpcEndpointRef): ApplicationInfo = {
  val now = System.currentTimeMillis()
  val date = new Date(now)
  val appId = newApplicationId(date)  // 生成 app-20260718-0000 格式的 ID
  new ApplicationInfo(now, appId, desc, date, driver, defaultCores)
}

registerApplication 将 Application 加入管理集合：

// Master.scala
private def registerApplication(app: ApplicationInfo): Unit = {
  val appAddress = app.driver.address
  // 防止重复注册
  if (addressToApp.contains(appAddress)) {
    logInfo("Attempted to re-register application at same address: " + appAddress)
    return
  }

  applicationMetricsSystem.registerSource(app.appSource)
  apps += app                       // 所有应用集合
  idToApp(app.id) = app             // ID → ApplicationInfo
  endpointToApp(app.driver) = app   // Endpoint → ApplicationInfo
  addressToApp(appAddress) = app    // 地址 → ApplicationInfo
  waitingApps += app                // 等待调度的应用队列
}

6.2 schedule() 调度入口

schedule() 是 Master 的资源调度入口。每次有新应用注册、新 Worker 加入、资源变化时都会调用：

// Master.scala
private def schedule(): Unit = {
  if (state != RecoveryState.ALIVE) {
    return  // 非 ALIVE 状态不调度
  }

  // Drivers take strict precedence over executors
  val shuffledAliveWorkers = Random.shuffle(
    workers.toSeq.filter(_.state == WorkerState.ALIVE))
  val numWorkersAlive = shuffledAliveWorkers.size
  var curPos = 0

  // 调度等待中的 Drivers（Round-Robin 轮询 Worker）
  // 从 curPos 位置开始轮询，每次尝试下一个 Worker，直到找到可用资源或遍历完所有 Worker
  // 每个 Driver 启动后，curPos 继续推进，保证多个 Driver 间公平分配
  for (driver <- waitingDrivers.toList) {
    var launched = false
    var numWorkersVisited = 0
    while (numWorkersVisited < numWorkersAlive && !launched) {
      val worker = shuffledAliveWorkers(curPos)
      numWorkersVisited += 1
      if (canLaunchDriver(worker, driver.desc)) {
        val allocated = worker.acquireResources(driver.desc.resourceReqs)
        driver.withResources(allocated)
        launchDriver(worker, driver)
        waitingDrivers -= driver
        launched = true
      }
      curPos = (curPos + 1) % numWorkersAlive
    }
  }

  // ② 调度 Executors
  startExecutorsOnWorkers()
}

Round-Robin 轮询机制详解：

curPos 是跨 Driver 持续累加的，不重置回 0：

• Driver A 尝试 Worker[0]（资源不足）→ Worker[1]（成功启动，curPos=2）
• Driver B 从 Worker[2] 开始尝试 → Worker[3]（成功启动，curPos=4）

这种”从上次停下的位置继续”的方式，保证多个 Driver 间公平分配 Worker，不会出现所有 Driver 都挤在同一个 Worker 上的情况。

为什么 Client 模式不走 Driver 调度？

关键在 schedule() 遍历的是 waitingDrivers 队列，而 Client 模式发送的是 RegisterApplication 消息，Master 处理时创建 ApplicationInfo 并加入 waitingApps。Driver 进入 waitingDrivers 的唯一途径是 Master 收到提交 Driver 的请求（Cluster 模式的 SubmitDriverRequest）。

Client 模式：RegisterApplication → ApplicationInfo → waitingApps  → startExecutorsOnWorkers()
Cluster 模式：SubmitDriverRequest → DriverInfo → waitingDrivers → launchDriver() + startExecutorsOnWorkers()

所以 Client 模式下 waitingDrivers 始终为空，Driver 调度循环直接跳过，走到 Executor 调度。

1. Drivers 优先调度（Round-Robin 方式遍历 Worker）
2. Executors 随后调度（通过 startExecutorsOnWorkers()）

6.3 startExecutorsOnWorkers() — Executor 调度

这是 Standalone 模式下 Executor 资源分配的核心方法：

// Master.scala

/**
 * 调度并启动 Worker 上的 Executors
 * 当前是一个简单的 FIFO 调度器
 */
private def startExecutorsOnWorkers(): Unit = {
  // 按 FIFO 顺序遍历等待队列中的 Application
  for (app <- waitingApps) {
    val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1)
    // 如果剩余核心数不够启动一个 Executor，跳过
    if (app.coresLeft >= coresPerExecutor) {
      // ① 过滤出可用的 Worker（ALIVE 状态且资源充足）
      val usableWorkers = workers.toArray
        .filter(_.state == WorkerState.ALIVE)       // 只选择存活 Worker
        .filter(canLaunchExecutor(_, app.desc))     // 资源满足需求
        .sortBy(_.coresFree).reverse                // 按空闲核心数降序排列

      // ② 计算每个 Worker 上分配的核心数
      val assignedCores = scheduleExecutorsOnWorkers(
        app, usableWorkers, spreadOutApps)

      // ③ 根据分配结果，在每个 Worker 上启动 Executor
      for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) {
        allocateWorkerResourceToExecutors(
          app, assignedCores(pos), app.desc.coresPerExecutor, usableWorkers(pos))
      }
    }
  }
}

核心步骤：

1. 按 FIFO 顺序遍历 waitingApps 队列
2. 过滤可用 Worker：存活状态、资源充足
3. 按空闲核心数降序排序：优先使用资源充足的 Worker
4. 计算分配方案：通过 scheduleExecutorsOnWorkers() 决定每个 Worker 分配多少核心
5. 执行分配：通过 allocateWorkerResourceToExecutors() 在 Worker 上启动 Executor

6.4 scheduleExecutorsOnWorkers() — 核心分配算法

这个方法实现具体的资源分配逻辑，支持两种模式：分散（spreadOut） 和 集中（非 spreadOut）：

// Master.scala

private def scheduleExecutorsOnWorkers(
    app: ApplicationInfo,
    usableWorkers: Array[WorkerInfo],
    spreadOutApps: Boolean): Array[Int] = {

  val coresPerExecutor = app.desc.coresPerExecutor    // 每个 Executor 的核心数
  val minCoresPerExecutor = coresPerExecutor.getOrElse(1)
  val oneExecutorPerWorker = coresPerExecutor.isEmpty  // 未配置时，每个 Worker 只启动一个 Executor
  val memoryPerExecutor = app.desc.memoryPerExecutorMB
  val numUsable = usableWorkers.length
  val assignedCores = new Array[Int](numUsable)       // 每个 Worker 分配的核心数
  val assignedExecutors = new Array[Int](numUsable)    // 每个 Worker 上的新 Executor 数
  var coresToAssign = math.min(app.coresLeft,
    usableWorkers.map(_.coresFree).sum)               // 待分配的总核心数

  /**
   * 判断指定 Worker 能否为当前 App 启动 Executor
   */
  def canLaunchExecutorForApp(pos: Int): Boolean = {
    val keepScheduling = coresToAssign >= minCoresPerExecutor
    val enoughCores = usableWorkers(pos).coresFree - assignedCores(pos) >= minCoresPerExecutor
    val assignedExecutorNum = assignedExecutors(pos)
    val launchingNewExecutor = !oneExecutorPerWorker || assignedExecutorNum == 0

    if (launchingNewExecutor) {
      val assignedMemory = assignedExecutorNum * memoryPerExecutor
      val enoughMemory = usableWorkers(pos).memoryFree - assignedMemory >= memoryPerExecutor
      // ... 检查资源需求 ...
      val underLimit = assignedExecutors.sum + app.executors.size < app.executorLimit
      keepScheduling && enoughCores && enoughMemory && enoughResources && underLimit
    } else {
      // 已有一个 Executor，只增加核心数
      keepScheduling && enoughCores
    }
  }

  // 循环分配核心，直到无法继续
  var freeWorkers = (0 until numUsable).filter(canLaunchExecutorForApp)
  while (freeWorkers.nonEmpty) {
    freeWorkers.foreach { pos =>
      var keepScheduling = true
      while (keepScheduling && canLaunchExecutorForApp(pos)) {
        coresToAssign -= minCoresPerExecutor
        assignedCores(pos) += minCoresPerExecutor

        if (oneExecutorPerWorker) {
          assignedExecutors(pos) = 1    // 每 Worker 一个 Executor
        } else {
          assignedExecutors(pos) += 1   // 可启动多个 Executor
        }

        // spreadOut 模式：每个 Worker 只分配一次，然后跳到下一个 Worker
        if (spreadOutApps) {
          keepScheduling = false
        }
      }
    }
    freeWorkers = freeWorkers.filter(canLaunchExecutorForApp)
  }
  assignedCores
}

spreadOut 算法说明：

• spreadOut = true（默认）：将 Executor 尽可能分散到多个 Worker 上，有利于数据本地性
• spreadOut = false：尽可能让 Executor 集中到尽可能少的 Worker 上
• 可通过 spark.deploy.spreadOut 配置（默认 true）

6.5 allocateWorkerResourceToExecutors() 与 launchExecutor()

分配确定后，allocateWorkerResourceToExecutors() 在具体 Worker 上启动 Executor：

// Master.scala

private def allocateWorkerResourceToExecutors(
    app: ApplicationInfo,
    assignedCores: Int,
    coresPerExecutor: Option[Int],
    worker: WorkerInfo): Unit = {

  // 计算要在该 Worker 上启动的 Executor 数量
  val numExecutors = coresPerExecutor.map { assignedCores / _ }.getOrElse(1)
  val coresToAssign = coresPerExecutor.getOrElse(assignedCores)

  for (i <- 1 to numExecutors) {
    val allocated = worker.acquireResources(app.desc.resourceReqsPerExecutor)
    val exec = app.addExecutor(worker, coresToAssign, allocated)
    launchExecutor(worker, exec)           // 发送 LaunchExecutor 到 Worker
    app.state = ApplicationState.RUNNING
  }
}

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.resources))

  // ② 向 AppClient 发送 ExecutorAdded 通知
  exec.application.driver.send(
    ExecutorAdded(exec.id, worker.id, worker.hostPort, exec.cores, exec.memory))
}

关键动作：

1. Master 向 Worker 发送 LaunchExecutor 消息，让 Worker 启动 Executor 进程
2. Master 向 AppClient 发送 ExecutorAdded 通知，告知 Driver 有新的 Executor 被分配

7. Worker 启动 Executor

Worker 收到 Master 的 LaunchExecutor 消息后，创建 ExecutorRunner 并启动：

7.1 Worker.receive — LaunchExecutor

// Worker.scala — receive 方法

case LaunchExecutor(masterUrl, appId, execId, appDesc, cores_, memory_, resources_) =>
  if (masterUrl != activeMasterUrl) {
    logWarning("Invalid Master (" + masterUrl + ") attempted to launch executor.")
  } else if (decommissioned) {
    logWarning("Asked to launch an executor while decommissioned. Not launching executor.")
  } else {
    try {
      logInfo("Asked to launch executor %s/%d for %s".format(appId, execId, appDesc.name))

      // ① 创建 Executor 的工作目录
      val executorDir = new File(workDir, appId + "/" + execId)
      if (!executorDir.mkdirs()) {
        throw new IOException("Failed to create directory " + executorDir)
      }

      // ② 创建 Executor 本地目录
      val appLocalDirs = appDirectories.getOrElse(appId, { ... })
      appDirectories(appId) = appLocalDirs

      // ③ 创建 ExecutorRunner——管理 Executor 进程的执行
      val manager = new ExecutorRunner(
        appId, execId,
        appDesc.copy(command = Worker.maybeUpdateSSLSettings(appDesc.command, conf)),
        cores_, memory_, self, workerId,
        webUi.scheme, host, webUi.boundPort, publicAddress,
        sparkHome, executorDir, workerUri, conf,
        appLocalDirs, ExecutorState.LAUNCHING, resources_)

      // ④ 缓存并启动
      executors(appId + "/" + execId) = manager
      manager.start()                              // ← 启动子进程
      coresUsed += cores_
      memoryUsed += memory_
      addResourcesUsed(resources_)
    } catch {
      case e: Exception =>
        logError(s"Failed to launch executor $appId/$execId for ${appDesc.name}.", e)
        // ... 清理并通知 Master
    }
  }

7.2 ExecutorRunner.start() 与 fetchAndRunExecutor()

ExecutorRunner.start() 在新线程中启动 fetchAndRunExecutor()，真正执行 CoarseGrainedExecutorBackend 进程：

// ExecutorRunner.scala

private[worker] def start(): Unit = {
  workerThread = new Thread("ExecutorRunner for " + fullId) {
    override def run(): Unit = { fetchAndRunExecutor() }
  }
  workerThread.start()
  // 注册 ShutdownHook
  shutdownHook = ShutdownHookManager.addShutdownHook { () =>
    if (state == ExecutorState.LAUNCHING || state == ExecutorState.RUNNING) {
      state = ExecutorState.FAILED
    }
    killProcess(Some("Worker shutting down"))
  }
}

fetchAndRunExecutor() 替换占位符并启动 CoarseGrainedExecutorBackend 进程：

// ExecutorRunner.scala

/**
 * 下载并运行我们的应用程序描述中描述的 executor
 */
private def fetchAndRunExecutor(): Unit = {
  try {
    val resourceFileOpt = prepareResourcesFile(SPARK_EXECUTOR_PREFIX, resources, executorDir)

    // ① 构建实际参数（替换占位符）
    val arguments = appDesc.command.arguments ++ resourceFileOpt.map(f =>
      Seq("--resourcesFile", f.getAbsolutePath)).getOrElse(Seq.empty)

    // ② 替换 {{EXECUTOR_ID}}、{{HOSTNAME}} 等占位符
    val subsOpts = appDesc.command.javaOpts.map {
      Utils.substituteAppNExecIds(_, appId, execId.toString)
    }
    val subsCommand = appDesc.command.copy(arguments = arguments, javaOpts = subsOpts)

    // ③ 构建进程
    val builder = CommandUtils.buildProcessBuilder(subsCommand, new SecurityManager(conf),
      memory, sparkHome.getAbsolutePath, substituteVariables)
    val command = builder.command()

    logInfo(s"Launch command: $redactedCommand")

    // ④ 设置环境变量
    builder.directory(executorDir)
    builder.environment.put("SPARK_EXECUTOR_DIRS", appLocalDirs.mkString(File.pathSeparator))
    builder.environment.put("SPARK_LAUNCH_WITH_SCALA", "0")

    // ⑤ 启动进程 —— 真正的 CoarseGrainedExecutorBackend 子进程
    process = builder.start()

    // ⑥ 重定向 stdout/stderr 到日志文件
    val stdout = new File(executorDir, "stdout")
    stdoutAppender = FileAppender(process.getInputStream, stdout, conf, true)

    val stderr = new File(executorDir, "stderr")
    Files.write(header, stderr, StandardCharsets.UTF_8)
    stderrAppender = FileAppender(process.getErrorStream, stderr, conf, true)

    // ⑦ 更新状态为 RUNNING，通知 Worker
    state = ExecutorState.RUNNING
    worker.send(ExecutorStateChanged(appId, execId, state, None, None))

    // ⑧ 等待进程退出
    val exitCode = process.waitFor()
    state = ExecutorState.EXITED
    val message = "Command exited with code " + exitCode
    worker.send(ExecutorStateChanged(appId, execId, state, Some(message), Some(exitCode)))
  } catch {
    case interrupted: InterruptedException =>
      state = ExecutorState.KILLED
      killProcess(None)
    case e: Exception =>
      state = ExecutorState.FAILED
      killProcess(Some(e.toString))
  }
}

占位符替换方法：

// ExecutorRunner.scala

/** 替换命令参数中的占位符 */
private[worker] def substituteVariables(argument: String): String = argument match {
  case "{{WORKER_URL}}" => workerUrl
  case "{{EXECUTOR_ID}}" => execId.toString
  case "{{HOSTNAME}}" => host
  case "{{CORES}}" => cores.toString
  case "{{APP_ID}}" => appId
  case other => other
}

启动后的进程命令大致如下：

/bin/java -server ...
  org.apache.spark.executor.CoarseGrainedExecutorBackend
  --driver-url spark://CoarseGrainedScheduler@driverHost:driverPort
  --executor-id 0
  --hostname workerHost
  --cores 4
  --app-id app-20260718-0000
  --worker-url spark://Worker@workerHost:workerPort

这些参数的含义和构建过程见 4.1 节”阶段二（Worker 端）”。

8. Executor 注册到 Driver

8.1 CoarseGrainedExecutorBackend 启动

CoarseGrainedExecutorBackend 进程启动后，执行 main() 方法，解析命令行参数并调用 run()：

// CoarseGrainedExecutorBackend.scala

def main(args: Array[String]): Unit = {
  // 解析命令行参数
  val createFn: (RpcEnv, Arguments, SparkEnv, ResourceProfile) =>
    CoarseGrainedExecutorBackend = { case (rpcEnv, arguments, env, resourceProfile) =>
    new CoarseGrainedExecutorBackend(rpcEnv, arguments.driverUrl, arguments.executorId,
      arguments.bindAddress, arguments.hostname, arguments.cores, arguments.userClassPath.toSeq,
      env, arguments.resourcesFileOpt, resourceProfile)
  }
  // 核心方法
  run(parseArguments(args, this.getClass.getCanonicalName.stripSuffix("$")), createFn)
  System.exit(0)
}

run() 方法执行 Executor 启动的完整流程：

// CoarseGrainedExecutorBackend.scala

def run(
    arguments: Arguments,
    backendCreateFn: (RpcEnv, Arguments, SparkEnv, ResourceProfile) =>
      CoarseGrainedExecutorBackend): Unit = {

  Utils.initDaemon(log)

  SparkHadoopUtil.get.runAsSparkUser { () =>

    val executorConf = new SparkConf
    // ① 创建临时 RpcEnv（"driverPropsFetcher"），连接到 Driver 获取 SparkConf 配置
    //    先连接 Driver 获取配置再创建正式 RpcEnv，原因是 Executor 进程启动时还不知道
    //    Spark 配置（这些配置在 Driver 端），需要先拉取过来才能创建正确的 SparkEnv
    val fetcher = RpcEnv.create("driverPropsFetcher", arguments.bindAddress,
      arguments.hostname, -1, executorConf, new SecurityManager(executorConf),
      numUsableCores = 0, clientMode = true)

    // 连接 Driver（重试 3 次）
    var driver: RpcEndpointRef = null
    val nTries = 3
    for (i <- 0 until nTries if driver == null) {
      try {
        driver = fetcher.setupEndpointRefByURI(arguments.driverUrl)
      } catch { case e: Throwable => if (i == nTries - 1) throw e }
    }

    // ② 向 Driver 请求 SparkAppConfig（包含 SparkConf、appId、加密密钥等）
    val cfg = driver.askSync[SparkAppConfig](RetrieveSparkAppConfig(arguments.resourceProfileId))
    val props = cfg.sparkProperties ++ Seq[(String, String)](("spark.app.id", arguments.appId))
    // 临时 RpcEnv 用完即关
    fetcher.shutdown()

    // ③ 根据从 Driver 获取的配置，创建正式的 SparkEnv
    val driverConf = new SparkConf()
    for ((key, value) <- props) { driverConf.set(key, value) }
    driverConf.set(EXECUTOR_ID, arguments.executorId)
    val env = SparkEnv.createExecutorEnv(driverConf, arguments.executorId, arguments.bindAddress,
      arguments.hostname, arguments.cores, cfg.ioEncryptionKey, isLocal = false)

    // ④ 创建 CoarseGrainedExecutorBackend 实例
    val backend = backendCreateFn(env.rpcEnv, arguments, env, cfg.resourceProfile)

    // ⑤ 注册 RPC Endpoint → 触发 onStart()（开始反向注册）
    env.rpcEnv.setupEndpoint("Executor", backend)

    // ⑥ 启动 WorkerWatcher，监控与 Worker 的连接状态
    arguments.workerUrl.foreach { url =>
      env.rpcEnv.setupEndpoint("WorkerWatcher",
        new WorkerWatcher(env.rpcEnv, url, isChildProcessStopping = backend.stopping))
    }

    // 阻塞等待退出
    env.rpcEnv.awaitTermination()
  }
}

8.2 onStart() — 反向注册

setupEndpoint 注册 CoarseGrainedExecutorBackend 到 RpcEnv 后，自动触发 onStart() 方法，开始向 Driver 反向注册：

// CoarseGrainedExecutorBackend.scala

override def onStart(): Unit = {
  logInfo("Connecting to driver: " + driverUrl)

  // 异步连接 Driver 的 DriverEndpoint
  rpcEnv.asyncSetupEndpointRefByURI(driverUrl).flatMap { ref =>
    driver = Some(ref)
    // 向 Driver 发送 RegisterExecutor 请求
    ref.ask[Boolean](RegisterExecutor(
      executorId, self, hostname, cores,
      extractLogUrls, extractAttributes,
      _resources, resourceProfile.id))
  }(ThreadUtils.sameThread).onComplete {
    case Success(_) =>
      // ask 返回 true，说明注册成功
      // 通过 self.send(RegisteredExecutor) 给自己发消息
      // 目的：把创建 Executor 的逻辑从异步回调线程切回 RPC 主线程，保证线程安全
      self.send(RegisteredExecutor)
    case Failure(e) =>
      exitExecutor(1, s"Cannot register with driver: $driverUrl", e, notifyDriver = false)
  }(ThreadUtils.sameThread)
}

8.3 DriverEndpoint 处理 RegisterExecutor

Driver 端的 DriverEndpoint 收到 RegisterExecutor 消息后：

// CoarseGrainedSchedulerBackend.scala — DriverEndpoint.receiveAndReply

case RegisterExecutor(executorId, executorRef, hostname, cores, logUrls,
     attributes, resources, resourceProfileId) =>

  if (executorDataMap.contains(executorId)) {
    // ① 重复注册 → 拒绝
    context.sendFailure(new IllegalStateException(s"Duplicate executor ID: $executorId"))
  } else if (scheduler.excludedNodes.contains(hostname) || ...) {
    // ② 节点被排除 → 拒绝
    context.sendFailure(...)
  } else {
    // ③ 记录 Executor 地址映射
    addressToExecutorId(executorAddress) = executorId
    totalCoreCount.addAndGet(cores)           // 累加总核数
    totalRegisteredExecutors.addAndGet(1)     // 累加注册数

    // ④ 创建 ExecutorData 存入 executorDataMap
    val data = new ExecutorData(executorRef, executorAddress, hostname,
      0, cores, logUrls, attributes, resources, resourceProfileId, ...)
    executorDataMap.put(executorId, data)

    // ⑤ 发布 Executor 添加事件（让监听者如 WebUI 知道）
    listenerBus.post(
      SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data))

    // ⑥ 回复注册成功（ask 的返回结果）
    //    CoarseGrainedExecutorBackend 收到 true 后，触发 self.send(RegisteredExecutor)
    //    将创建 Executor 对象的逻辑从异步回调切回 RPC 主线程（见 8.2 节）
    context.reply(true)
  }

8.4 创建 Executor 实例

CoarseGrainedExecutorBackend 收到 RegisteredExecutor 消息后，创建 Executor 实例：

// CoarseGrainedExecutorBackend.scala — receive 方法

override def receive: PartialFunction[Any, Unit] = {
  case RegisteredExecutor =>
    logInfo("Successfully registered with driver")
    // 注册成功后，创建真正的 Executor 计算引擎
    executor = new Executor(executorId, hostname, env, userClassPath, isLocal = false, resources)
}

至此，CoarseGrainedExecutorBackend 进程 完成了注册（8.2），进程内部创建了 Executor 实例（executor = new Executor(...)），可以开始接收并执行 Task 了。

关于 Executor 这个名称的歧义说明

本文及 Spark 文档中多处出现”Executor”，实际指两个不同层次的概念：

概念	本质	对应代码
Executor 进程	独立 JVM 进程	CoarseGrainedExecutorBackend.main() 启动的进程
Executor 实例	进程内部的 Task 执行与 RDD 缓存对象	new Executor(executorId, ...)

• 业界通用的 “Spark Executor”、UI 界面上显示的 Executor 条目、本文第 7~8 章讨论的注册过程，都是指 Executor 进程
• 第 8.4 节 new Executor(...) 特指进程内部的 Executor 对象，是实现细节

不同模式下 Executor 进程的入口类有所不同：

模式	Executor 进程入口类	说明
Standalone	CoarseGrainedExecutorBackend	Spark 自带调度系统，直接启动子进程
Yarn	YarnCoarseGrainedExecutorBackend	继承自 CoarseGrainedExecutorBackend，适配 YARN Container 生命周期
Kubernetes	CoarseGrainedExecutorBackend	与 Standalone 相同入口类，通过 K8s 启动 Pod

但无论哪种模式，Executor 进程内部最终都通过 new Executor(...) 创建实例，Task 执行逻辑完全一致。

关系：Worker/Container 启动 Executor 进程 → 进程向 Driver 注册 → 注册成功后进程内部创建 Executor 实例 → 实例执行 Task

9. 总结

流程概述

Standalone Client 模式的完整提交流程：

1. SparkSubmit 提交程序 → prepareSubmitEnvironment 设置 childMainClass = args.mainClass → runMain 在客户端反射执行用户主类的 main 方法，Driver 在客户端启动
2. SparkContext 初始化 → 创建 TaskSchedulerImpl + StandaloneSchedulerBackend
3. StandaloneSchedulerBackend.start() → 创建 DriverEndpoint → 构建 ApplicationDescription → 创建 StandaloneAppClient
4. StandaloneAppClient 向 Master 发送 RegisterApplication（重试 3 次，间隔 20 秒）
5. Master 处理注册 → 创建 ApplicationInfo → 回复 RegisteredApplication → 调用 schedule() 调度资源
6. Master.schedule() → startExecutorsOnWorkers() → FIFO 遍历 waitingApps → 过滤可用 Worker → allocateWorkerResourceToExecutors() → launchExecutor() 发送 LaunchExecutor 到 Worker
7. Worker 收到 LaunchExecutor → 创建 ExecutorRunner → fetchAndRunExecutor() → 启动子进程 CoarseGrainedExecutorBackend（即 Executor 进程）
8. CoarseGrainedExecutorBackend 进程 启动 → run() 先通过临时 RpcEnv 连接 Driver 获取 SparkConf 配置 → 创建正式 SparkEnv → 注册 RPC Endpoint 触发 onStart() → 发送 RegisterExecutor → Driver 确认注册 → 进程内部创建 Executor 实例（new Executor(...)）
9. Driver 继续执行用户业务代码 → 创建 RDD/DataFrame → Action 算子触发 Job → DAGScheduler/TaskScheduler 调度 Task 到已注册的 Executor 上执行

Standalone Client vs Yarn Client 对比

特性	Standalone Client	Yarn Client
Driver 位置	客户端	客户端
集群管理器	Master + Worker	ResourceManager + NodeManager
资源调度	Master.schedule() 自研调度器	YARN 框架调度
Executor 入口类	CoarseGrainedExecutorBackend	YarnCoarseGrainedExecutorBackend
Executor 启动方式	Worker 直接启动子进程	NMClient 启动 Container
AM/中间角色	无（Client 直连 Master）	ExecutorLauncher（AM）
Executor 资源申请	Master 主动分配	YarnAllocator 向 RM 申请
注册流程	StandaloneAppClient → Master → Worker	YarnClientSchedulerBackend → RM → AM → NM
重试机制	3 次，每次 20 秒	YARN 框架重试
适用场景	Spark 自建集群，轻量级	已有 Hadoop/YARN 集群，企业生产

核心调用链

SparkSubmit.main
  → submit → runMain
    → 反射调用用户主类.main（Driver 在客户端启动）
      → SparkContext 初始化 → _taskScheduler.start()
        → StandaloneSchedulerBackend.start()                          [4]
          → super.start()
          → 构建 Executor 启动命令 (CoarseGrainedExecutorBackend)    [4.1]
          → 创建 ApplicationDescription                              [4.2]
          → new StandaloneAppClient(rpcEnv, masters, appDesc, ...)
          → client.start()                                           [5]
            → setupEndpoint("AppClient", ClientEndpoint)
              → ClientEndpoint.onStart()
                → registerWithMaster(1)                              [5.2]
                  → tryRegisterAllMasters()
                    → 向所有 Master 发送 RegisterApplication

        [Master 集群侧]
        Master.receive: RegisterApplication                          [6.1]
          → createApplication() + registerApplication()
          → persistenceEngine.addApplication(app)
          → driver.send(RegisteredApplication)
          → schedule()                                               [6.2]
            → startExecutorsOnWorkers()                              [6.3]
              → filter workers, sort by free cores
              → scheduleExecutorsOnWorkers()                         [6.4]
              → allocateWorkerResourceToExecutors()                  [6.5]
                → launchExecutor()
                  → worker.endpoint.send(LaunchExecutor)

        [Worker 集群侧]
        Worker.receive: LaunchExecutor                               [7.1]
          → new ExecutorRunner().start()                             [7.2]
            → fetchAndRunExecutor()
              → 替换占位符 {{EXECUTOR_ID}}, {{HOSTNAME}} 等
              → builder.start()
                → CoarseGrainedExecutorBackend.main()

        [Executor 进程启动]
        CoarseGrainedExecutorBackend.main                            [8.1]
          → parseArguments → run()
            → 临时 RpcEnv 连接 Driver 获取 SparkConf
            → 创建正式 SparkEnv
            → setupEndpoint("Executor", backend)
              → onStart()                                              [8.2]
                → 向 DriverEndpoint 发送 RegisterExecutor
                  → Driver 回复成功                                    [8.3]
                    → self.send(RegisteredExecutor)
                      → new Executor(...)                              [8.4]

        [Driver 执行业务代码]
        DAGScheduler.runJob
          → handleJobSubmitted → createResultStage
            → submitStage → submitMissingTasks
              → taskScheduler.submitTasks

完整的流程图

┌─────────────────────────────────────────────────────────────────────────┐
│ 提交客户端 (Client)                                                     │
│                                                                         │
│  SparkSubmit.main                                                       │
│    │                                                                    │
│    ├─ runMain 反射调用用户主类.main                                     │
│    │    │                                                               │
│    │    ├─ SparkContext 初始化                                           │
│    │    │    │                                                          │
│    │    │    ├─ TaskSchedulerImpl + StandaloneSchedulerBackend          │
│    │    │    │    │                                                     │
│    │    │    │    ├─ super.start()                                     │
│    │    │    │    ├─ 构建 Executor 命令                                 │
│    │    │    │    ├─ ApplicationDescription                             │
│    │    │    │    ├─ StandaloneAppClient                                │
│    │    │    │    │    │                                               │
│    │    │    │    │    ├─ onStart()                                    │
│    │    │    │    │    │    └─ registerWithMaster(1)                   │
│    │    │    │    │    │         └─ RegisterApplication ────────────────┼──→ Master
│    │    │    │    │    │                                               │    │
│    │    │    │    │    ├─ receive(RegisteredApplication)               │    ├─ createApplication()
│    │    │    │    │    │    └─ listener.connected(appId)              │    ├─ registerApplication()
│    │    │    │    │    │         └─ registrationBarrier.release()     │    ├─ schedule()
│    │    │    │    │    │                                              │    │    │
│    │    │    │    │    ├─ waitForRegistration() 返回                  │    │    ├─ startExecutorsOnWorkers()
│    │    │    │    │    │                                              │    │    │    │
│    │    │    │    │    ├─ [Executor 注册完成后]                       │    │    │    └─ allocateWorkerResourceToExecutors()
│    │    │    │    │    │    └─ receive(ExecutorAdded)                 │    │    │         │
│    │    │    │    │    │                                              │    │    │         └─ launchExecutor()
│    │    │    │    │    │                                              │    │    │              │
│    │    │    │    │    │                                              │    │    │              ├─ Worker: LaunchExecutor
│    │    │    │    │    │                                              │    │    │              └─ Client: ExecutorAdded
│    │    │    │    │    │                                              │    │    │
│    │    │    │    │    └─ [继续执行业务代码]                          │    │    │
│    │    │    │    │         ├─ 创建 RDD/DataFrame                    │    │    │
│    │    │    │    │         └─ Action 算子触发 Job                    │    │    │
│    │    │    │    │              ├─ DAGScheduler 划分 Stage           │    │    │
│    │    │    │    │              └─ TaskScheduler 调度 Task 到        │    │    │
│    │    │    │    │                  已注册 Executor                  │    │    │
│    │    │    │    │                                                  │    │    │
│    │    │    │    └─ DriverEndpoint ←──────────────────────────────────────┼────┼── Executor
│    │    │    │         ↑  RegisterExecutor                           │    │    │    ↑
│    │    │    │         ↓  RegisteredExecutor                         │    │    │    │
│    │    │    │                                                       │    │    │    │
│    │    │    │                                                       │    │    │    ├─ CoarseGrainedExecutorBackend
│    │    │    │                                                       │    │    │    │    │
│    │    │    │                                                       │    │    │    │    ├─ onStart()
│    │    │    │                                                       │    │    │    │    │  └─ RegisterExecutor
│    │    │    │                                                       │    │    │    │    │       → DriverEndpoint
│    │    │    │                                                       │    │    │    │    │
│    │    │    │                                                       │    │    │    │    ├─ receive(RegisteredExecutor)
│    │    │    │                                                       │    │    │    │    │  └─ new Executor(...)
│    │    │    │                                                       │    │    │    │    │
│    │    │    │                                                       │    │    │    │    └─ [接收并执行 Task]
│    │    │    │                                                       │    │    │    │
│    │    │    │                                                       │    │    │    └─ Worker (Node)
│    │    │    │                                                       │    │    │         │
│    │    │    │                                                       │    │    │         ├─ receive(LaunchExecutor)
│    │    │    │                                                       │    │    │         ├─ ExecutorRunner.start()
│    │    │    │                                                       │    │    │         └─ fetchAndRunExecutor()
│    │    │    │                                                       │    │    │              └─ builder.start()
│    │    │    │                                                       │    │    │                   → CoarseGrainedExecutorBackend
└─────────────────────────────────────────────────────────────────────────┘    │    │
                                                                               │    │
                                                        Master (集群) ◄────────┘    │
                                                         │                        │
                                                         ├─ 管理 Workers            │
                                                         ├─ 接收应用注册           │
                                                         ├─ 调度 Executor          │
                                                         └─ 状态管理               │
                                                                                  │
                                                              Worker (集群) ◄──────┘
                                                               │
                                                               ├─ 注册到 Master
                                                               ├─ 接收 LaunchExecutor
                                                               ├─ 启动 Executor 进程
                                                               └─ 监控 Executor 状态

括号中的 [x.x] 对应本文各章节编号，便于快速回溯源码分析。

下一次我们将分析 Standalone Cluster 模式的详细提交流程，以及它和 Client 模式在 Driver 启动、资源调度等方面的差异。

赏

感谢您的支持!

支付宝

微信

本文由 董可伦 发表于 伦少的博客 ,采用署名-非商业性使用-禁止演绎 3.0进行许可。

非商业转载请注明作者及出处。商业转载请联系作者本人。

本文标题：Spark 源码 | Standalone Client 模式提交流程（七）

本文链接：https://dongkelun.com/2026/07/23/spark/sparkSourceCode-standalone-client/

欢迎关注我的公众号