Spark 源码 | Yarn Client 模式提交流程（五）

2026-07-12

前言

在第四篇文章 SparkContext 初始化与 TaskScheduler 创建流程（四）中，我们分析了 SparkContext 初始化和 TaskScheduler 创建流程。本文详细介绍 Yarn Client 模式的提交流程。

版本

Spark 3.2.3

Yarn Client 模式概述

Yarn Client 模式下，Driver 运行在提交应用的客户端机器上，直接与 YARN ResourceManager 通信申请资源，Executor 运行在 YARN 的 NodeManager 节点上。

什么是 Driver？Driver 为什么运行在客户端？

在 Spark 中，Driver 并不对应某个单独的类或对象，而是一个运行时角色——用户主类的 main 方法在哪里执行，Driver 就在哪里（用户主类中会创建 SparkContext/SparkSession，驱动整个应用的执行）。

第二篇文章 SparkSubmit 提交流程分析（二）中，prepareSubmitEnvironment 方法根据 deployMode 设置 childMainClass，决定了 main 方法在何处执行：

模式	childMainClass	main 方法执行位置	Driver 位置
Yarn Client	`args.mainClass`（用户主类）	`runMain` 在客户端反射调用	客户端
Yarn Cluster	`YarnClusterApplication`	`runDriver` 在 AM 内调用	集群

总结：Yarn Client 模式 childMainClass = args.mainClass，本地反射执行用户主类的 main 方法，Driver 在客户端；Yarn Cluster 模式 childMainClass = YarnClusterApplication，由 ApplicationMaster 通过 runDriver() 启动用户主类，Driver 在集群中。

关于 Driver 及各模式下角色的详细总结，将在系列文章完成后单独分析。

1 2	// 提交命令示例 spark-submit --master yarn --deploy-mode client --class org.apache.spark.examples.SparkPi $SPARK_HOME/examples/jars/spark-examples_2.12-3.2.3.jar

more >>

展开全文 >>

Spark 源码 | SparkContext 初始化与 TaskScheduler 创建流程（四）

2026-07-09

前言

在第二篇文章 Spark 源码 | SparkSubmit 提交流程分析（二）中，我们分析了 SparkSubmit 的提交流程。在 Yarn Client 模式下：

runMain 方法会调用 app.start（Yarn Client 模式下为 JavaMainApplication），在 app.start 中通过反射调用用户主类的 main 方法（具体调用过程在不同模式中有所不同，后续分析各模式时再详细说明）
用户主类（如 SparkPi）的 main 方法通常会创建 SparkSession 或 SparkContext
在创建 SparkContext 时，会创建 TaskScheduler 和 SchedulerBackend

本文说明：虽然本文以 Yarn Client 模式为例（YarnScheduler、YarnClientSchedulerBackend），但 SparkContext 初始化和 TaskScheduler 创建流程是各模式的公共前提。后续分析 Yarn Cluster、Standalone 等模式时可直接参考本篇内容。

为什么要先写这篇：TaskScheduler 和 SchedulerBackend 的创建是各模式的公共前提，内容较多。如果不先单独分析，在分析各模式提交流程时就需要同时包含这些内容，导致每篇文章都很长。因此本文先专门分析它们的创建过程，后续分析各模式时就可以直接聚焦于该模式特有的流程（如 Executor 启动）。

版本

Spark 3.2.3

SparkContext 初始化入口

用户代码通常这样创建 SparkContext：

val spark = SparkSession.builder()
  .appName("WordCount")
  .master("yarn")
  .getOrCreate()

val sc = spark.sparkContext

more >>

展开全文 >>

Spark 源码 | SparkSubmitArguments 参数解析（三）

2026-07-08

前言

在第二篇文章 Spark 源码 | SparkSubmit 提交流程分析（二）中，我们分析了 SparkSubmit 的提交流程，其中 doSubmit 方法会调用 parseArguments 解析命令行参数。本文详细介绍参数解析的过程。

版本

Spark 3.2.3

SparkSubmitArguments 概述

SparkSubmitArguments 继承自 SparkSubmitArgumentsParser，用于解析 spark-submit 命令行参数。

1
2
3

// SparkSubmitArguments.scala
private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, String] = sys.env)
  extends SparkSubmitArgumentsParser with Logging {

more >>

展开全文 >>

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

2026-07-06

前言

在上一篇文章 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()
  }
}

more >>

展开全文 >>

Flink CDC2Kafka 总结

2026-04-22

前言

本文主要记录 CDC2Kafka 使用过程中的一些细节配置，以 mysql-cdc 为例。

背景需求

之前 CDC 的主要使用场景并没有写到 Kafka ，而是直接 Sink 到支持更新的目标端，比如关系型数据库、Hudi、HBase，现在有了写到 Kafka 的需求，主要原因是有的项目的厂商不会使用数据湖表格式(Hudi、Iceberg、Delta Lake)，在使用 Hive 时不能支持ACID ，那么只能通过覆盖分区的方式实现更新删除，而覆盖更新对比小表来说可以通过定时全量抽取实现，对于大表每次都全量抽取效率太慢，所以最好可以通过增量变更日志+历史Hive表数据合并来实现，对于这种需求就得通过 CDC 格式来实现，CDC 格式有 debezium-json、canal-json、maxwell-json、ogg-json、changelog-json。关系型数据库、Hudi、HBase等是不支持CDC格式的，常用的比较熟悉的组件就是Kafka，另外后面发现直接写 HDFS 文件也是可以的，本文先总结 CDC2Kafka+debezium-json 中的一些具体需求配置。

debezium-json

格式数据示例：

{"before":{"id":3,"name":"hudi3","price":3.33,"ts":2000,"dt":"20230331"},"after":null,"op":"d"}
{"before":null,"after":{"id":3,"name":"hudi4","price":3.33,"ts":2000,"dt":"20230331"},"op":"c"}

{"before":null,"after":{"id":4,"name":"hudi4","price":3.33,"ts":2000,"dt":"20230331"},"op":"c"}

{"before":{"id":4,"name":"hudi4","price":3.33,"ts":2000,"dt":"20230331"},"after":null,"op":"d"}

分别对应update、insert、delete ，其中 update 一次产生两条记录，先删除再插入。

more >>

展开全文 >>