夜雨聆风前言
在第二篇文章 Spark 源码 | SparkSubmit 提交流程分析（二） 中，我们分析了 SparkSubmit 的提交流程，其中 doSubmit 方法会调用 parseArguments 解析命令行参数。本文详细介绍参数解析的过程。
版本
Spark 3.2.3
SparkSubmitArguments 概述
SparkSubmitArguments 继承自 SparkSubmitArgumentsParser，用于解析 spark-submit 命令行参数。
// SparkSubmitArguments.scalaprivate[deploy] classSparkSubmitArguments(args: Seq[String], env: Map[String, String] = sys.env)extendsSparkSubmitArgumentsParserwithLogging {字段定义
// --master MASTER_URL// Spark Master 地址，如 spark://host:port, mesos://host:port, yarn, k8s://https://host:port, local（默认 local[*]）var master: String = null// --deploy-mode DEPLOY_MODE// 部署模式，client 表示在本地启动 Driver，cluster 表示在集群的 Worker 节点上启动 Driver（默认 client）var deployMode: String = null// --executor-memory MEM// 每个 Executor 的内存（如 1000M, 2G）（默认 1G）var executorMemory: String = null// --executor-cores NUM// 每个 Executor 的 CPU 核心数（YARN 和 K8S 模式默认 1，Standalone 模式默认为 Worker 上的所有核心）var executorCores: String = null// --total-executor-cores NUM// 所有 Executor 的总 CPU 核心数var totalExecutorCores: String = null// --properties-file FILE// 配置文件路径，用于加载额外的 Spark 配置。如果未指定，则默认查找 conf/spark-defaults.confvar propertiesFile: String = null// --driver-memory MEM// Driver 的内存（如 1000M, 2G）var driverMemory: String = null// --driver-class-path// 额外添加到 Driver classpath 的路径。注意：通过 --jars 添加的 jar 包会自动包含在 classpath 中var driverExtraClassPath: String = null// --driver-library-path// 额外添加到 Driver 库路径的环境变量var driverExtraLibraryPath: String = null// --driver-java-options// 传递给 Driver 的额外 Java 选项var driverExtraJavaOptions: String = null// --queue QUEUE_NAME// YARN 队列名称（默认 "default"）var queue: String = null// --num-executors NUM// Executor 的数量（默认 2）。启用动态分配时，实际初始数量取 --num-executors、// spark.dynamicAllocation.minExecutors 和 spark.dynamicAllocation.initialExecutors 三者中的最大值var numExecutors: String = null// --files FILES// 逗号分隔的文件列表，会被放置在每个 Executor 的工作目录中。在 Executor 中可通过 SparkFiles.get(fileName) 获取文件路径var files: String = null// --archives ARCHIVES// 逗号分隔的归档文件列表，会被提取到每个 Executor 的工作目录中var archives: String = null// --class CLASS_NAME// 应用程序的主类（用于 Java / Scala 应用）var mainClass: String = null// <app jar | python file | R file>// 第一个未被识别的选项被视为"主资源"（primary resource）var primaryResource: String = null// --name NAME// 应用程序的名称var name: String = null// 应用程序参数var childArgs: ArrayBuffer[String] = newArrayBuffer[String]()// --jars JARS// 逗号分隔的 JAR 包列表，会被添加到 Driver 和 Executor 的 classpath 中var jars: String = null// --packages// 逗号分隔的 Maven 坐标列表，用于添加 JAR 包到 Driver 和 Executor 的 classpath。// 会先搜索本地 Maven 仓库，然后是 Maven 中央仓库，以及 --repositories 指定的远程仓库。// 坐标格式为 groupId:artifactId:versionvar packages: String = null// --repositories// 逗号分隔的远程仓库地址，用于搜索 --packages 指定的 Maven 坐标var repositories: String = nullvar ivyRepoPath: String = nullvar ivySettingsPath: Option[String] = None// --exclude-packages// 逗号分隔的 groupId:artifactId 列表，在解析 --packages 提供的依赖时排除这些包，以避免依赖冲突var packagesExclusions: String = null// --verbose, -v// 打印额外的调试输出var verbose: Boolean = falsevar isPython: Boolean = false// --py-files PY_FILES// 逗号分隔的 .zip、.egg 或 .py 文件列表，会被添加到 Python 应用的 PYTHONPATH 中var pyFiles: String = nullvar isR: Boolean = falsevar action: SparkSubmitAction = null// 通过 `--conf` 和配置文件加载的 Spark 配置属性val sparkProperties: HashMap[String, String] = newHashMap[String, String]()// --proxy-user NAME// 提交应用程序时模拟的用户名。不能与 --principal / --keytab 同时使用var proxyUser: String = null// --principal PRINCIPAL// 用于登录 KDC 的 Principal（仅 YARN/K8s 模式）var principal: String = null// --keytab KEYTAB// 包含上述 Principal 对应 keytab 的文件路径var keytab: String = nullprivatevar dynamicAllocationEnabled: Boolean = false// --supervise// 如果设置，Driver 失败时会自动重启（仅 Spark Standalone 或 Mesos 的 cluster 部署模式）var supervise: Boolean = false// --driver-cores NUM// Driver 使用的 CPU 核心数，仅在 cluster 模式可用（默认 1）var driverCores: String = null// --kill SUBMISSION_ID// 如果设置，杀死指定的 Driver（Spark Standalone、Mesos 或 K8s 的 cluster 部署模式）var submissionToKill: String = null// --status SUBMISSION_ID// 如果设置，查询指定 Driver 的状态var submissionToRequestStatusFor: String = null// 内部使用，用于 REST APIvar useRest: Boolean = false解析流程
1. 构造函数调用顺序
// Set parameters from command line arguments// 从命令行参数设置参数parse(args.asJava)// Populate `sparkProperties` map from properties file// 从属性文件填充 sparkPropertiesmergeDefaultSparkProperties()// Remove keys that don't start with "spark." from `sparkProperties`.// 从 sparkProperties 中移除不以 "spark." 开头的键ignoreNonSparkProperties()// Use `sparkProperties` map along with env vars to fill in any missing parameters// 使用 sparkProperties 和环境变量填充缺失的参数loadEnvironmentArguments()useRest = sparkProperties.getOrElse("spark.master.rest.enabled", "false").toBooleanvalidateArguments()解析顺序：
1. parse(args.asJava) - 从命令行参数设置参数 2. mergeDefaultSparkProperties() - 从属性文件填充 sparkProperties 3. ignoreNonSparkProperties() - 从 sparkProperties 中移除不以 "spark." 开头的键 4. loadEnvironmentArguments() - 使用 sparkProperties 和环境变量填充缺失的参数 5. validateArguments() - 验证参数
2. 解析命令行参数
继承自 SparkSubmitArgumentsParser（继承自 SparkSubmitOptionParser），重写 handle、handleUnknown、handleExtraArgs 三个回调方法处理参数。
SparkSubmitOptionParser parse 方法解析逻辑
parse(List<String> args) 方法的遍历逻辑：
// 遍历所有参数for (idx = 0; idx < args.size(); idx++) {// 支持 --master=spark://xxx 格式（等号分隔）// 检查是否是 --xxx=value 格式，若是则拆分Matcherm= eqSeparatedOpt.matcher(arg);if (m.matches()) { arg = m.group(1); value = m.group(2); }// Look for options with a value.// 第一步：在 opts 数组中查找（带参数的选项，如 --master, --class 等）Stringname= findCliOption(arg, opts);if (name != null) {// If the option requires a value, fetch it.// 命令行格式：--选项 值（分开的两参数），如 --master yarnif (value == null) {if (idx == args.size() - 1) {thrownewIllegalArgumentException( String.format("Missing argument for option '%s'.", arg)); } idx++; value = args.get(idx); }if (!handle(name, value)) {// handle 返回 false 的情况：--version（--help 会直接退出，不会返回）break; }continue; }// Look for a switch.// 第二步：在 switches 数组中查找（不带参数的选项，如 --help, --verbose 等） name = findCliOption(arg, switches);if (name != null) {// handle 返回 false 的情况：--version（打印版本后退出）if (!handle(name, null)) {break; }continue; }// 第三步：既不在 opts 也不在 switches 中，调用 handleUnknown 处理未知选项// SparkSubmitArguments 中将第一个未知选项作为 primaryResource，handleUnknown 始终返回 false// 返回 false 则 break 退出循环，剩余参数作为应用参数if (!handleUnknown(arg)) {break; }}// 跳过第一个未知选项（primaryResource），只把剩余参数传给 handleExtraArgs// 如果正常遍历完（无 break），idx == args.size()，不需要 idx++// 如果因 break 提前退出，idx < args.size()，需要跳过 primaryResourceif (idx < args.size()) { idx++;}// 第四步：解析完成后，剩余的参数作为应用参数调用 handleExtraArgs// 注意：循环可能因 break 提前退出，导致 idx < args.size()// 如：handleUnknown 返回 false 时（SparkSubmitArguments 将第一个未知选项作为 primaryResource）// spark-submit 命令格式：spark-submit [options] <app jar | python file> [app arguments]// 剩余的参数即 [app arguments]，如 arg1 arg2 arg3handleExtraArgs(args.subList(idx, args.size()));SparkSubmitArguments 重写的方法
1. handle 方法
处理已知的带参数选项，根据 opt（选项名）将 value 赋值给对应的字段。其中 CONF case 会直接将 --conf 的值存入 sparkProperties：
overrideprotecteddefhandle(opt: String, value: String): Boolean = { opt match {caseNAME => name = valuecaseMASTER => master = valuecaseCLASS => mainClass = valuecaseDEPLOY_MODE =>if (value != "client" && value != "cluster") { error("--deploy-mode must be either \"client\" or \"cluster\"") } deployMode = valuecaseNUM_EXECUTORS => numExecutors = value// ... 其他参数// `--conf` KEY=VALUE 格式的配置项，直接存入 sparkProperties// 由于是直接赋值（覆盖写），后续合并配置文件时，`--conf` 的值会被保留caseCONF =>val (confName, confValue) = SparkSubmitUtils.parseSparkConfProperty(value) sparkProperties(confName) = confValue// ... 其他参数 }// --version 时返回 false，其他情况返回 true action != SparkSubmitAction.PRINT_VERSION}2. handleUnknown 方法
处理未知选项，将第一个未知选项作为 primaryResource（主资源），如 myapp.jar、spark-shell、hdfs:///path/to/app.jar：
overrideprotecteddefhandleUnknown(opt: String): Boolean = {// 如果选项以 "-" 开头但不在已知选项中，报错if (opt.startsWith("-")) { error(s"Unrecognized option '$opt'.") }// 设置 primaryResource（主资源），如 myapp.jar、spark-shell、hdfs:///path/to/app.jar// 如果不是 shell（如 spark-shell、pyspark-shell、sparkr-shell）或内部选项（spark-internal），则通过 Utils.resolveURI 解析为标准 URI 格式// 如：./app.jar -> file:/path/to/app.jar，/path/app.jar -> file:/path/app.jar，hdfs://xxx -> hdfs://xxx// 否则直接作为字符串（如 spark-shell 用于交互式界面，spark-internal 用于测试场景） primaryResource =if (!SparkSubmit.isShell(opt) && !SparkSubmit.isInternal(opt)) {Utils.resolveURI(opt).toString } else { opt }// 判断是否为 Python 或 R 应用 isPython = SparkSubmit.isPython(opt) isR = SparkSubmit.isR(opt)// 始终返回 false，退出循环false}3. handleExtraArgs 方法
将解析剩余的应用参数添加到 childArgs：
overrideprotecteddefhandleExtraArgs(extra: JList[String]): Unit = {// 在 SparkSubmitArguments 类中只有这一处赋值，没看到有其它地方赋值，感觉两种写法效果可能是相同的// 使用 ++= 追加而非 = 赋值，可能为了和 SparkSubmit.scala 中对 childArgs 的操作风格保持一致// childArgs 是 ArrayBuffer[String]，extra.asScala 是 Seq[String]// 使用 ++= 避免了显式类型转换（如 childArgs = extra.asScala.to[ArrayBuffer]） childArgs ++= extra.asScala}3. 合并默认属性文件
将配置文件（默认 conf/spark-defaults.conf）中的属性合并到 sparkProperties 中。如果用户通过 --conf 指定了相同的配置项，则保留 --conf 的值（优先级更高）。
privatedefmergeDefaultSparkProperties(): Unit = {// Use common defaults file, if not specified by user// 如果用户未指定，则使用默认属性文件 propertiesFile = Option(propertiesFile).getOrElse(Utils.getDefaultPropertiesFile(env))// Honor `--conf` before the defaults file// `--conf` 的优先级高于默认属性文件 defaultSparkProperties.foreach { case (k, v) =>if (!sparkProperties.contains(k)) { sparkProperties(k) = v } }}4. 过滤非 Spark 属性
将 sparkProperties 中不以 spark. 开头的键移除掉。sparkProperties 中的值来自 --conf 和配置文件，不包含环境变量等来源。
privatedefignoreNonSparkProperties(): Unit = { sparkProperties.keys.foreach { k =>if (!k.startsWith("spark.")) { sparkProperties -= k logWarning(s"Ignoring non-Spark config property: $k") } }}5. 加载环境变量参数
将命令行未指定的参数从环境变量和 sparkProperties 中补充进来。例如：master 未在命令行指定时，会依次从 spark.master 配置、环境变量 MASTER 中获取。
privatedefloadEnvironmentArguments(): Unit = {// Load arguments from environment variables, Spark properties etc.// 从环境变量、Spark 属性等加载参数 master = Option(master) .orElse(sparkProperties.get("spark.master")) .orElse(env.get("MASTER")) .orNull// ... 其他参数类似}优先级：命令行参数 > sparkProperties > 环境变量 > 默认值
6. 参数验证
验证必填参数是否存在以及参数值是否合法。例如：必须指定 primaryResource（主资源），内存/核心数必须为正数，YARN 模式必须设置 HADOOP_CONF_DIR 等。
privatedefvalidateSubmitArguments(): Unit = {// Ensure that required fields exists. Call this only once all defaults are loaded.// 确保必填字段存在。调用此方法时所有默认值已加载完成if (args.length == 0) { printUsageAndExit(-1) }if (primaryResource == null) { error("Must specify a primary resource (JAR or Python or R file)") }// ... 其他验证}参数优先级
--conf | |
总结
SparkSubmitArguments 的解析流程：
1. parse - 解析命令行参数 2. mergeDefaultSparkProperties - 合并配置文件中的属性 3. ignoreNonSparkProperties - 过滤非 spark. 开头的属性 4. loadEnvironmentArguments - 用环境变量和 sparkProperties 填充空缺参数 5. validateArguments - 验证参数合法性
优先级：命令行 > --conf > 配置文件 > 环境变量 > 默认值
下一次我们将分析 Yarn Client、Yarn Cluster、Standalone Client、Standalone Cluster 模式的详细提交流程。
🧐 分享、点赞、在看，给个3连击呗！👇
