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

展开全文 >>

Flink DataStream2Table 总结

2026-03-04

前言

平时工作中有时会遇到 DataStream 转 Table 的需求，之前也写过几次，但是不总结就会忘掉还得专门扒拉之前写的代码，比较麻烦，现在总结一下，方便后面再有需要时查找。

DataStream(Tuple)

DataStream<Tuple> 转 Table

DataStream<Tuple5<Integer, String, Double, Long, String>> dataStream = env.fromCollection(
        java.util.Arrays.asList(
                Tuple5.of(1, "手机", 5999.99, 1733400000000L, "2025-12-05"),
                Tuple5.of(2, "电脑", 8999.50, 1733400001000L, "2025-12-05"),
                Tuple5.of(3, "耳机", 399.00, 1733400002000L, "2025-12-05")
        )
);

// 字段顺序必须和 Tuple5 一致：id(第1位)、name(第2位)、price(第3位)、ts(第4位)、dt(第5位)
Table table = tableEnv.fromDataStream(
        dataStream,
        "id, name, price, ts, dt" // 直接指定字段名，类型会自动匹配（Tuple5 的泛型）
);

table.execute().print();    
table.printSchema();

more >>

展开全文 >>

Hive 集群安装配置

2025-11-06

前言

接Zookeeper 安装配置和 Hadoop HA 集群安装配置，本文继续总结 Hive 集群安装配置。其实之前的文章中已经总结过部分，但是不全面：

Centos7 hive 单机模式安装配置：单机版元数据库初始化、Hive CLI 使用嵌入式 Metastore 模式连接 Hive 。
Spark连接Hive（spark-shell和Eclipse两种方式）：单机版独立 Metastore 服务启动、Spark 通过独立的 HMS 服务连接 Hive 。
通过数据库客户端界面工具DBeaver连接Hive: 单机版 HiveServer2 启动(使用嵌入式 Metastore)、DBeaver 通过 HiveServer2 JBDCUrl 地址连接Hive 。

前提

主机名修改、SSH互信、JDK安装等。

版本

Hive 3.1.3

安装包

https://archive.apache.org/dist/hive/hive-3.1.3/apache-hive-3.1.3-bin.tar.gz

more >>

展开全文 >>

Ray 集群安装配置

2025-09-26

前言

作为新手，学习总结一下 Ray 的集群安装配置及简单使用。

官网

英文：https://docs.ray.io/en/latest/index.html
中文：https://docs.rayai.org.cn/en/latest/index.html
介绍
Ray 是一个开源的分布式计算框架，旨在简化分布式应用的开发和部署。它提供了简单易用的 API，让开发者能够轻松地将单机应用扩展到分布式环境，适用于机器学习、深度学习、数据处理等多种场景。

Ray 的核心特点

简单的分布式编程模型：通过装饰器（@ray.remote）可以轻松将函数或类转换为分布式版本，无需深入了解分布式系统细节。
自动任务调度：Ray 拥有高效的任务调度器，能够自动在集群中分配任务，优化资源利用率。
支持状态 ful 计算：除了无状态的任务，Ray 还支持有状态的 Actor 模型，适合需要维护状态的场景（如训练模型、服务部署）。
丰富的生态系统：包含多个上层库，如用于超参数调优的 Ray Tune、用于强化学习的 Ray RLlib、用于分布式训练的 Ray Train 等。
灵活的部署方式：可在单机、集群、云环境（AWS、GCP、Azure 等）中部署，支持动态资源扩展。 more >>

展开全文 >>

前言

版本

SparkSubmit 入口

前言

背景需求

debezium-json

前言

DataStream(Tuple)

前言

前提

版本

安装包

前言

官网

介绍

Ray 的核心特点