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

2026-07-18

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

前言

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

Standalone 模式是 Spark 自带的集群管理器，不依赖 YARN、Mesos 等外部资源调度系统。其架构包含 Master（集群管理器）和 Worker（工作节点）两个角色，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

完整提交流程

1. SparkSubmit 准备阶段

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

对于 Standalone Client 模式，关键代码在 SparkSubmit.scala 的 prepareSubmitEnvironment 方法中：

// SparkSubmit.scala — prepareSubmitEnvironment 方法

// ① 判断集群管理器：master 以 "spark" 开头 → STANDALONE
val clusterManager: Int = args.master match {
  case "yarn" => YARN
  case m if m.startsWith("spark") => STANDALONE   // Standalone 模式
  case m if m.startsWith("mesos") => MESOS
  case m if m.startsWith("k8s") => KUBERNETES
  case m if m.startsWith("local") => LOCAL
  case _ => error(...)
}

// ② 判断 deployMode：默认 client
val deployMode: Int = args.deployMode match {
  case "client" | null => CLIENT
  case "cluster" => CLUSTER
  case _ => error(...)
}

集群管理器识别后，进入 Standalone 模式的配置赋值逻辑。当 deployMode == CLIENT 时，设置 childMainClass：

// SparkSubmit.scala — prepareSubmitEnvironment 方法（第 777-783 行）

// 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 方法中的典型调用链：

1
2
3

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

3. SparkContext 初始化与 TaskScheduler 创建

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

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

TaskScheduler：TaskSchedulerImpl
SchedulerBackend：StandaloneSchedulerBackend

// SparkContext.scala — createTaskScheduler 方法（简化）
private def createTaskScheduler(
    sc: SparkContext,
    master: String): (SchedulerBackend, TaskScheduler) = {

  import SparkContext._

  master match {
    case "spark://..." =>  // Standalone 模式
      val scheduler = new TaskSchedulerImpl(sc)
      val backend = new StandaloneSchedulerBackend(scheduler, sc, urls)
      scheduler.initialize(backend)
      (backend, scheduler)

    case "yarn" =>
      // YARN 模式：YarnScheduler + YarnClientSchedulerBackend
      ...
  }
}

注意：这个模式匹配只是一个简化示意。实际源码使用了 ExternalClusterManager 的 SPI 机制来创建，但最终效果如上所述。

创建完成后，SparkContext 初始化过程中会调用 _taskScheduler.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()
    //    创建 DriverEndpoint（RPC 端点），用于与 Executor 通信
    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 super.start() — CoarseGrainedSchedulerBackend.start()

CoarseGrainedSchedulerBackend.start() 创建 DriverEndpoint，是 Driver 端与 Executor 通信的 RPC 端点：

// CoarseGrainedSchedulerBackend.scala
override def start(): Unit = {
  // ① 创建 DriverEndpoint 并注册到 RpcEnv
  val driverEndpoint = rpcEnv.setupEndpoint(
    CoarseGrainedSchedulerBackend.ENDPOINT_NAME,
    new DriverEndpoint(rpcEnv, scheduler, this))

  // ② 为每个 ResourceProfile 创建任务调度器辅助对象
  ...
}

DriverEndpoint 负责处理 Executor 的注册（RegisterExecutor）、任务分发（LaunchTask）、状态更新（StatusUpdate）等核心 RPC 消息。

4.2 启动命令模板与 CoarseGrainedExecutorBackend

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

1	org.apache.spark.executor.CoarseGrainedExecutorBackend

与 YARN 模式有所不同：

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

CoarseGrainedExecutorBackend 的启动参数使用了占位符，Worker 在真正启动 Executor 进程时会替换为实际值：

val args = Seq(
  "--driver-url", driverUrl,          // ← 实际地址，如 spark://CoarseGrainedScheduler@host:port
  "--executor-id", "{{EXECUTOR_ID}}", // ← 占位符，由 Worker 替换
  "--hostname", "{{HOSTNAME}}",       // ← 占位符
  "--cores", "{{CORES}}",             // ← 占位符
  "--app-id", "{{APP_ID}}",           // ← 占位符
  "--worker-url", "{{WORKER_URL}}")   // ← 占位符

4.3 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 注册流程

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() 方法，开始注册流程：

// 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 地址发起注册

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)
      }
    })
  }
}

注意：使用 send（异步、无需响应）而不是 ask（同步、需响应）。Master 注册成功后，会通过 RegisteredApplication 消息回复。

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()          // 释放信号量 → waitForRegistration() 返回
  launcherBackend.setAppId(appId)
}

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 严格优先于 Executors 调度
  val shuffledAliveWorkers = Random.shuffle(
    workers.toSeq.filter(_.state == WorkerState.ALIVE))
  val numWorkersAlive = shuffledAliveWorkers.size
  var curPos = 0

  // 调度等待中的 Drivers（Round-Robin）
  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()
}

调度优先级：

Drivers 优先调度（Round-Robin 方式遍历 Worker）

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))
      }
    }
  }
}

核心步骤：

按 FIFO 顺序遍历 waitingApps 队列
过滤可用 Worker：存活状态、资源充足
按空闲核心数降序排序：优先使用资源充足的 Worker
计算分配方案：通过 scheduleExecutorsOnWorkers() 决定每个 Worker 分配多少核心
执行分配：通过 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))
}

关键动作：

Master 向 Worker 发送 LaunchExecutor 消息，让 Worker 启动 Executor 进程

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

8. Executor 注册到 Driver

8.1 CoarseGrainedExecutorBackend 启动

CoarseGrainedExecutorBackend 进程启动后，执行 main() 方法：

// CoarseGrainedExecutorBackend.scala

object CoarseGrainedExecutorBackend {

  def main(args: Array[String]): Unit = {
    // 解析命令行参数
    val createFn: (RpcEnv, SparkConf, String, String, Map[String, String], Int, Int, Int, String) =>
      CoarseGrainedExecutorBackend = { ... }

    // 核心方法
    run(driverUrl, executorId, hostname, cores, appId, workerUrl, conf, createFn)
  }

  private def run(
      driverUrl: String,
      executorId: String,
      hostname: String,
      cores: Int,
      appId: String,
      workerUrl: String,
      conf: SparkConf,
      createFn: (RpcEnv, SparkConf, String, String, Map[String, String], Int, Int, Int, String)
        => CoarseGrainedExecutorBackend): Unit = {

    // ① 创建 RpcEnv
    val env = RpcEnv.create("Executor", hostname, -1, conf, securityMgr)

    // ② 连接到 Driver，获取 SparkConf 配置
    val driverRef = env.setupEndpointRefByURI(driverUrl)
    val cfg = driverRef.askSync[SparkAppConfig](RetrieveSparkAppConfig(
      resourceProfileId))

    // ③ 创建 SparkEnv
    val envInstance = SparkEnv.createExecutorEnv(conf, executorId, hostname, cores,
      cfg.ioEncryptionKey, isLocal = false)

    // ④ 创建 CoarseGrainedExecutorBackend 实例
    val backend = createFn(envInstance, cfg, driverUrl, executorId, hostname,
      cores, cfg.resourceProfileId)

    // ⑤ 注册 RPC Endpoint → 触发 onStart()
    envInstance.rpcEnv.setupEndpoint("Executor", backend)

    // 阻塞等待退出
    backend.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(_) =>
      // 注册成功后，发消息给自己，触发创建 Executor
      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

override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {

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

    // ① 检查是否已注册
    if (executorDataMap.contains(executorId)) {
      // 已注册 → 重新发送 RegisteredExecutor
      context.reply(true)
    } else {
      // ② 记录 Executor 信息
      val data = new ExecutorData(executorRef, executorRef.address, hostname,
        cores, cores, logUrls, attributes, resources, resourceProfileId)
      CoarseGrainedSchedulerBackend.this.synchronized {
        executorDataMap.put(executorId, data)   // 缓存 Executor 数据
        if (currentExecutorIdCounter < executorId.toInt) {
          currentExecutorIdCounter = executorId.toInt
        }
      }

      // ③ 回复注册成功
      context.reply(true)

      // ④ 通知 TaskScheduler 有新的 Executor 可用
      scheduler.postStartHook()

      // ⑤ 发送 RegisteredExecutor 确认到 Executor（可选，取决于是否有未完成的任务）
      ...
    }
}

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（进程）完成了注册，内部创建了 Executor（计算对象），可以开始接收任务了。

9. 总结

流程概述

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

SparkSubmit 提交程序 → prepareSubmitEnvironment 设置 childMainClass = args.mainClass → runMain 在客户端反射执行用户主类的 main 方法，Driver 在客户端启动
SparkContext 初始化 → 创建 TaskSchedulerImpl + StandaloneSchedulerBackend
StandaloneSchedulerBackend.start() → 创建 DriverEndpoint → 构建 ApplicationDescription → 创建 StandaloneAppClient
StandaloneAppClient 向 Master 发送 RegisterApplication（重试 3 次，间隔 20 秒）
Master 处理注册 → 创建 ApplicationInfo → 回复 RegisteredApplication → 调用 schedule() 调度资源
Master.schedule() → startExecutorsOnWorkers() → FIFO 遍历 waitingApps → 过滤可用 Worker → allocateWorkerResourceToExecutors() → launchExecutor() 发送 LaunchExecutor 到 Worker
Worker 收到 LaunchExecutor → 创建 ExecutorRunner → fetchAndRunExecutor() → 启动子进程 CoarseGrainedExecutorBackend
CoarseGrainedExecutorBackend 启动 → 连接 DriverEndpoint → 发送 RegisterExecutor → 注册成功后创建 Executor 对象
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() 创建 DriverEndpoint                        [4.1]
          → 构建 Executor 启动命令 (CoarseGrainedExecutorBackend)    [4.2]
          → 创建 ApplicationDescription                              [4.3]
          → 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]
          → create RpcEnv + 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() → DriverEndpoint                    │
│    │    │    │    ├─ 构建 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/18/spark/sparkSourceCode-standalone-client/

欢迎关注我的公众号