Spark 源码 | SparkSubmit 提交流程分析（二）

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

欢 迎 关 注 我 的 公 众 号

前言

在上一篇文章 Spark 源码 | 脚本分析总结（一） 中，我们分析了 Spark 的脚本执行流程，最终都通过 spark-class 启动 org.apache.spark.deploy.SparkSubmit，本文我们继续分析 SparkSubmit 的提交流程，看看它是如何根据不同的 clusterManager 和 deployMode 来提交应用的。

版本

Spark 3.2.3

SparkSubmit 入口

根据脚本分析，我们知道 spark-submit 最终通过 spark-class 启动 org.apache.spark.deploy.SparkSubmit，入口方法为 main：

// SparkSubmit.scala
override def main(args: Array[String]): Unit = {
  val submit = new SparkSubmit() {
    self =>
    // ...
  }
  submit.doSubmit(args)
}

doSubmit 方法是真正的入口：

def doSubmit(args: Array[String]): Unit = {
  // 初始化日志
  val uninitLog = initializeLogIfNecessary(true, silent = true)
  // 解析参数
  val appArgs = parseArguments(args)
  // 根据 action 执行对应操作
  appArgs.action match {
    case SparkSubmitAction.SUBMIT => submit(appArgs, uninitLog)
    case SparkSubmitAction.KILL => kill(appArgs)
    case SparkSubmitAction.REQUEST_STATUS => requestStatus(appArgs)
    case SparkSubmitAction.PRINT_VERSION => printVersion()
  }
}

参数解析

parseArguments 方法创建 SparkSubmitArguments 对象，解析命令行参数，将在下一篇文章中单独分析

protected def parseArguments(args: Array[String]): SparkSubmitArguments = {
  new SparkSubmitArguments(args)
}

SparkSubmitArguments 解析后的参数包括（以 spark-shell 为例）：

master: yarn
deployMode: null (默认 client)
executorMemory: null
executorCores: null
totalExecutorCores: null
propertiesFile: /usr/hdp/current/spark-client/conf/spark-defaults.conf
driverMemory: null
driverCores: null
driverExtraClassPath: null
driverExtraLibraryPath: /usr/hdp/current/hadoop-client/lib/native:...
driverExtraJavaOptions: -agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005
queue: spark
numExecutors: null
files: null
pyFiles: null
archives: null
mainClass: org.apache.spark.repl.Main
primaryResource: spark-shell
name: Spark shell
childArgs: [-verbose]
jars: null
packages: null
repositories: null
packagesExclusions: null
verbose: false
proxyUser: null
principal: null
keytab: null
supervise: false

提交流程

1. submit 方法

submit 方法是提交流程的统一入口，主要处理：

1. Standalone Cluster 模式特殊处理：支持 REST API 提交，失败时回退到 Legacy 方式
2. 调用 doRunMain：内嵌函数，最终执行 runMain

调用链：submit → doRunMain → runMain → prepareSubmitEnvironment（prepareSubmitEnvironment 在 runMain 中调用）

/**
 * 提交应用的主入口方法。
 *
 * 主要功能：
 * 1. 处理代理用户（--proxy-user）切换
 * 2. Standalone Cluster 模式的 REST API 回退逻辑
 * 3. 调用 runMain 启动应用
 */
@tailrec
private def submit(args: SparkSubmitArguments, uninitLog: Boolean): Unit = {

  /**
   * 执行 runMain 的内部函数。
   * 如果指定了 --proxy-user，以代理用户身份执行。
   */
  def doRunMain(): Unit = {
    if (args.proxyUser != null) {
      // 代理用户处理：如果指定了 --proxy-user，以代理用户身份执行
      val proxyUser = UserGroupInformation.createProxyUser(args.proxyUser,
        UserGroupInformation.getCurrentUser())
      try {
        proxyUser.doAs(new PrivilegedExceptionAction[Unit]() {
          override def run(): Unit = {
            runMain(args, uninitLog)
          }
        })
      } catch {
        case e: Exception =>
          if (e.getStackTrace().length == 0) {
            error(s"ERROR: ${e.getClass().getName()}: ${e.getMessage()}")
          } else {
            throw e
          }
      }
    } else {
      runMain(args, uninitLog)
    }
  }

  // In standalone cluster mode, there are two submission gateways:
  //   (1) The traditional RPC gateway using o.a.s.deploy.Client as a wrapper
  //   (2) The new REST-based gateway introduced in Spark 1.3
  // The latter is the default behavior as of Spark 1.3, but Spark submit will fail over
  // to use the legacy gateway if the master endpoint turns out to be not a REST server.
  // Standalone Cluster 模式支持两种提交方式：RPC 和 REST API
  if (args.isStandaloneCluster && args.useRest) {
    try {
      logInfo("Running Spark using the REST application submission protocol.")
      doRunMain()
    } catch {
      // Fail over to use the legacy submission gateway
      // 如果 REST 失败，回退到 legacy 方式
      case e: SubmitRestConnectionException =>
        logWarning(s"Master endpoint ${args.master} was not a REST server. " +
          "Falling back to legacy submission gateway instead.")
        args.useRest = false
        submit(args, false)
    }
    // In all other modes, just run the main class as prepared
    // 其他所有模式（包括 Client 模式、Yarn Cluster 等）直接执行
  } else {
    doRunMain()
  }
}

2. prepareSubmitEnvironment 方法

这是核心方法，根据不同的 clusterManager 和 deployMode 准备提交环境，返回：

• childArgs: 运行主类的参数
• childClasspath: 主类的 jar
• sparkConf: Spark 配置
• childMainClass: 主类的全路径

2.1 解析 clusterManager 和 deployMode

// 设置集群管理器
val clusterManager: Int = args.master match {
  case "yarn" => YARN           // 1
  case m if m.startsWith("spark") => STANDALONE  // 2
  case m if m.startsWith("mesos") => MESOS       // 4
  case m if m.startsWith("k8s") => KUBERNETES    // 16
  case m if m.startsWith("local") => LOCAL       // 8
  case _ =>
    error("Master must either be yarn or start with spark, mesos, k8s, or local")
    -1
}

// 设置部署模式，默认 client
val deployMode: Int = args.deployMode match {
  case "client" | null => CLIENT  // 1
  case "cluster" => CLUSTER       // 2
  case _ =>
    error("Deploy mode must be either client or cluster")
    -1
}

2.2 计算模式标志

val isYarnCluster = clusterManager == YARN && deployMode == CLUSTER
val isMesosCluster = clusterManager == MESOS && deployMode == CLUSTER
val isStandAloneCluster = clusterManager == STANDALONE && deployMode == CLUSTER
val isKubernetesCluster = clusterManager == KUBERNETES && deployMode == CLUSTER
val isKubernetesClient = clusterManager == KUBERNETES && deployMode == CLIENT

2.3 Client 模式

Client 模式与 Cluster 模式的区别：

• Client 模式：直接启动用户传入的主类（args.mainClass）
• Cluster 模式：启动包装类，用户主类作为参数传入

例如 spark-shell 提交时，args.mainClass 为 org.apache.spark.repl.Main，详情可参考 Spark 源码 | 脚本分析总结（一）

// In client mode, childMainClass equals the user-provided main class directly
// 在客户端模式下，childMainClass 直接等于用户传入的主类
if (deployMode == CLIENT) {
  childMainClass = args.mainClass
  if (localPrimaryResource != null && isUserJar(localPrimaryResource)) {
    childClasspath += localPrimaryResource
  }
  if (localJars != null) { childClasspath ++= localJars.split(",") }
}

2.4 Yarn Cluster 模式

// In yarn-cluster mode, use yarn.Client as a wrapper around the user class
// yarn-cluster 通过 YarnClusterApplication 启动 driver
if (isYarnCluster) {
  childMainClass = YARN_CLUSTER_SUBMIT_CLASS  // "org.apache.spark.deploy.yarn.YarnClusterApplication"
  if (args.isPython) {
    childArgs += ("--primary-py-file", args.primaryResource)
    childArgs += ("--class", "org.apache.spark.deploy.PythonRunner")
  } else if (args.isR) {
    childArgs += ("--primary-r-file", mainFile)
    childArgs += ("--class", "org.apache.spark.deploy.RRunner")
  } else {
    if (args.primaryResource != SparkLauncher.NO_RESOURCE) {
      childArgs += ("--jar", args.primaryResource)
    }
    childArgs += ("--class", args.mainClass)
  }
  if (args.childArgs != null) {
    args.childArgs.foreach { arg => childArgs += ("--arg", arg) }
  }
}

Yarn Cluster 模式下，childMainClass 变为 org.apache.spark.deploy.yarn.YarnClusterApplication，用户的主类作为参数传入。

2.5 Standalone Cluster 模式

// In standalone cluster mode, use the REST client to submit the application (Spark 1.3+).
if (args.isStandaloneCluster) {
  if (args.useRest) {
    childMainClass = REST_CLUSTER_SUBMIT_CLASS  // "org.apache.spark.deploy.rest.RestSubmissionClientApp"
    childArgs += (args.primaryResource, args.mainClass)
  } else {
    // In legacy standalone cluster mode, use Client as a wrapper around the user class
    childMainClass = STANDALONE_CLUSTER_SUBMIT_CLASS  // "org.apache.spark.deploy.ClientApp"
    if (args.supervise) { childArgs += "--supervise" }
    Option(args.driverMemory).foreach { m => childArgs += ("--memory", m) }
    Option(args.driverCores).foreach { c => childArgs += ("--cores", c) }
    childArgs += "launch"
    childArgs += (args.master, args.primaryResource, args.mainClass)
  }
  if (args.childArgs != null) {
    childArgs ++= args.childArgs
  }
}

Standalone Cluster 模式有两种方式：

• REST API：使用 RestSubmissionClientApp
• Legacy：使用 ClientApp

3. runMain 方法

runMain 方法真正启动应用：

private def runMain(args: SparkSubmitArguments, uninitLog: Boolean): Unit = {
  // 获取准备好的环境
  val (childArgs, childClasspath, sparkConf, childMainClass) = prepareSubmitEnvironment(args)

  // 初始化类加载器
  val loader = getSubmitClassLoader(sparkConf)
  for (jar <- childClasspath) {
    addJarToClasspath(jar, loader)
  }

  // 加载主类
  var mainClass: Class[_] = null
  try {
    mainClass = Utils.classForName(childMainClass)
  } catch {
    case e: ClassNotFoundException =>
      logError(s"Failed to load class $childMainClass.")
      throw new SparkUserAppException(CLASS_NOT_FOUND_EXIT_STATUS)
    case e: NoClassDefFoundError =>
      logError(s"Failed to load $childMainClass: ${e.getMessage()}")
      throw new SparkUserAppException(CLASS_NOT_FOUND_EXIT_STATUS)
  }

  // 创建 SparkApplication 或 JavaMainApplication
  val app: SparkApplication = if (classOf[SparkApplication].isAssignableFrom(mainClass)) {
    mainClass.getConstructor().newInstance().asInstanceOf[SparkApplication]
  } else {
    new JavaMainApplication(mainClass)
  }

  // 启动应用
  try {
    app.start(childArgs.toArray, sparkConf)
  } catch {
    case t: Throwable =>
      throw findCause(t)
  }
}

关键逻辑：

1. 加载主类（childMainClass）
2. 如果主类实现了 SparkApplication 接口，直接实例化
3. 否则包装成 JavaMainApplication
4. 调用 app.start 启动应用

不同模式的 childMainClass

模式	childMainClass
Client (Yarn/Standalone/Mesos/K8s/Local)	用户传入的主类（如 org.apache.spark.repl.Main）
Yarn Cluster	org.apache.spark.deploy.yarn.YarnClusterApplication
Mesos Cluster	org.apache.spark.deploy.rest.RestSubmissionClientApp
Standalone Cluster (REST)	org.apache.spark.deploy.rest.RestSubmissionClientApp
Standalone Cluster (Legacy)	org.apache.spark.deploy.ClientApp
K8s Cluster	org.apache.spark.deploy.k8s.submit.KubernetesClientApplication

总结

通过本文的分析，我们了解了 SparkSubmit 的提交流程：

1. 入口：spark-submit → spark-class → SparkSubmit.main → doSubmit
2. 参数解析：parseArguments 解析命令行参数（将在下一篇文章中单独分析）
3. submit 方法：
   • 处理代理用户（proxyUser）切换
   • Standalone Cluster 模式支持 REST API，失败时回退到 Legacy
   • 注意：submit 本身不区分模式，所有模式都会调用 runMain
4. 环境准备：prepareSubmitEnvironment 根据 clusterManager + deployMode 设置 childMainClass 和 childArgs（这是真正的模式区分点）
5. 应用启动：runMain 加载 childMainClass 并启动

流程说明：

• submit 方法是统一的入口，无论 Client 还是 Cluster 模式，最终都会调用 runMain
• runMain 调用 prepareSubmitEnvironment，在其中根据 clusterManager + deployMode 决定：
  • Client 模式：childMainClass = 用户传入的主类（如 org.apache.spark.repl.Main）
  • Cluster 模式：childMainClass = 包装类（如 YarnClusterApplication），用户主类作为参数传入
• 最后加载 childMainClass 并启动

下一次我们将分析参数解析详细过程，然后分别分析 Yarn Client、Yarn Cluster、Standalone Client、Standalone Cluster 模式的详细提交流程。

赏

感谢您的支持!

支付宝

微信

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

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

本文标题：Spark 源码 | SparkSubmit 提交流程分析（二）

本文链接：https://dongkelun.com/2026/07/06/spark/sparkSourceCode-sparkSubmit/

欢迎关注我的公众号