Apache Hudi 学习笔记 07：Spark 集成 Hudi 实战

spark hudi data-lake lakehouse spark-sql
hudi
发布日期:   2025-08-05
文章字数:   6.6k
阅读时长:   31 分
阅读次数:  

前面几篇把 Hudi 的 Timeline、文件布局、索引、表类型和写入操作拆开讲了一遍。到这一篇，开始进入具体计算引擎集成。Spark 是最适合先学的一条线，因为它同时覆盖批处理、交互式 SQL、离线初始化、数据修复和 spark-submit 程序提交。

Spark 集成 Hudi 时，最容易混淆的是三件事：第一，Hudi 不是 Spark 自带能力，必须加载 Hudi Spark bundle；第二，Spark DataSource 和 Spark SQL 看起来入口不同，但底层都要走 Hudi 写入逻辑；第三，Hive Metastore、Spark SQL Catalog、Hudi Catalog 之间的关系不能混着配，否则很容易出现奇怪的兼容性问题。

1. Spark 集成基础

1.1 背景问题

假设有一条订单入湖链路：

业务系统 / CDC / 离线文件 -> Spark -> Hudi 表 -> Spark SQL / Hive / Trino 查询

如果只是写 Parquet，Spark 只要 df.write.parquet(path) 就够了。但写 Hudi 表时，Spark 还要把下面这些信息交给 Hudi：

信息 作用
record key 判断业务记录是否是同一条
partition path 决定数据落到哪个分区目录
precombine field 同一批相同 key 数据保留哪条
table type 决定 COW 或 MOR
operation 决定 bulk_insert、insert、upsert、delete 等写入语义
catalog / extension 让 Spark SQL 能识别 Hudi DDL 和 DML

所以 Spark 集成 Hudi 不是只加一个 jar，更重要的是把 Spark Session、Hudi bundle、Hadoop/HDFS、Catalog 和写入参数统一起来。

1.2 版本选择

这篇使用的版本如下：

组件 版本 选择原因 可用性说明
Apache Hudi 1.0.2 Hudi 1.0.x 系列较新，适合围绕新版表格式和 Spark 3.5 做测试 2025-08-05 本文环境已验证
Apache Spark 3.5.6 Hudi 1.0.2 提供 Spark 3.5 对应 bundle，Spark 3.5.6 适合作为本文稳定测试组合 2025-08-05 本文环境已验证
Scala 2.12 当前 Hudi Spark bundle 文件名为 _2.12，需要与 Spark 发行包匹配 spark-3.5.6-bin-hadoop3 对应 Scala 2.12
Hadoop 3.4.1 前面搭建的 HDFS/YARN 环境，Hudi 表数据写入 HDFS 2025-08-05 本文环境已验证
Java 11 当前 Spark/Hudi 组合下验证通过 2025-08-05 本文环境已验证

服务器上实际看到的 Hudi Spark bundle：

find /opt -name "hudi-spark*bundle*.jar" | sort

输出如下：

/opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar
/opt/module/spark-3.5.6-bin-hadoop3/jars/hudi-spark3.5-bundle_2.12-1.0.2.jar

这里要注意两个点：

  1. hudi-spark3.5-bundle_2.12-1.0.2.jar 对应 Spark 3.5 和 Scala 2.12。
  2. 如果使用 Spark 3.4 或 Spark 3.3，需要换成对应构建的 Hudi bundle，不能只改 Spark 目录。

1.3 Spark 集成 Hudi 的整体流程

下面这张图是我对 Spark 集成 Hudi 的理解。

Spark 写入 Hudi 流程

这张图主要说明 spark-shell、spark-sql、spark-submit、DataSource API 虽然入口不同，但最终都需要加载 Hudi Spark bundle，并进入 Hudi Writer / Reader。

Spark DataSource、Spark SQL、spark-submit 三个入口最终都会进入 Hudi Spark Writer。真正决定写入结果的不是入口形式，而是 Hudi 表配置和写入参数。

我一般这样区分：

入口 适合场景
Spark DataSource 程序化写入、离线任务、批量修复、和已有 Spark 作业集成
Spark SQL 交互式建表、调试、数据修复、验证 Hudi DML
spark-submit 生产任务提交，把 DataSource 或 SQL 逻辑封装成应用
spark-shell 学习、排错、快速验证 Hudi 参数

1.4 启动 spark-shell

启动 spark-shell 时需要加 Hudi bundle 和三个关键配置：

spark-shell \
  --master local[2] \
  --jars /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar

几个配置的作用如下：

配置 作用
--jars 把 Hudi Spark bundle 加到 Spark driver 和 executor classpath
spark.serializer 使用 Kryo，降低序列化开销
spark.sql.extensions 让 Spark SQL 支持 Hudi 的 SQL 扩展
spark.sql.catalog.spark_catalog 让 Spark 默认 Catalog 能处理 Hudi 表
spark.kryo.registrator 注册 Hudi 相关 Kryo 序列化类

我在服务器上启动后输出如下：

Spark context Web UI available at http://lavm-bzoq5mwl1h:4040
Spark context available as 'sc' (master = local[2], app id = local-1754383013367).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.5.6
      /_/

Using Scala version 2.12.18 (OpenJDK 64-Bit Server VM, Java 11.0.13)
Type in expressions to have them evaluated.
Type :help for more information.

scala> :quit

这类启动输出主要看 Spark 版本、Scala 版本、Java 版本、Spark Session 是否创建成功，以及 Hudi bundle 是否能被加载。外部访问地址在博客里没有必要展开，保留主机名和本地端口已经够用。

2. Spark 读写实践

2.1 Spark DataSource 写入 Hudi

Spark DataSource 是最常用的程序接入方式。下面用 PySpark 写一个 COW 表，路径为 /hudi/demo_spark_ds_orders。

DataSource 写入时，最关键的是把业务数据和 Hudi 写入参数一起交给 Hudi Spark Writer。整体链路如下：

Spark DataSource 写入 Hudi

这张图主要说明 DataSource 写入时，业务 DataFrame 会和 record key、partition path、precombine、operation 等 Hudi 参数一起交给 Hudi Spark Writer。

先准备脚本：

cat > /tmp/hudi_spark_datasource_demo.py <<'PY'
from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("hudi-spark-datasource-demo")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog")
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
    .config("spark.kryo.registrator", "org.apache.spark.HoodieSparkKryoRegistrar")
    .getOrCreate())

base_path = "/hudi/demo_spark_ds_orders"
table_name = "demo_spark_ds_orders"

spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration()).delete(
    spark._jvm.org.apache.hadoop.fs.Path(base_path), True
)

common_options = {
    "hoodie.table.name": table_name,
    "hoodie.datasource.write.recordkey.field": "order_id",
    "hoodie.datasource.write.partitionpath.field": "dt",
    "hoodie.datasource.write.precombine.field": "ts",
    "hoodie.datasource.write.table.type": "COPY_ON_WRITE",
    "hoodie.datasource.write.keygenerator.class": "org.apache.hudi.keygen.SimpleKeyGenerator",
    "hoodie.metadata.enable": "true"
}

columns = ["order_id", "user_id", "amount", "status", "ts", "dt"]

inserts = spark.createDataFrame([
    (7001, "u101", 99.8, "CREATED", 1754295600000, "2025-08-04"),
    (7002, "u102", 168.0, "CREATED", 1754295660000, "2025-08-04"),
    (7003, "u103", 39.9, "CREATED", 1754382000000, "2025-08-05")
], columns)

(inserts.write.format("hudi")
    .options(**common_options)
    .option("hoodie.datasource.write.operation", "bulk_insert")
    .mode("overwrite")
    .save(base_path))

updates = spark.createDataFrame([
    (7002, "u102", 168.0, "PAID", 1754296800000, "2025-08-04"),
    (7004, "u104", 258.6, "CREATED", 1754382060000, "2025-08-05")
], columns)

(updates.write.format("hudi")
    .options(**common_options)
    .option("hoodie.datasource.write.operation", "upsert")
    .mode("append")
    .save(base_path))

spark.read.format("hudi").load(base_path).createOrReplaceTempView("orders")

print("=== snapshot query ===")
spark.sql("""
SELECT order_id, user_id, amount, status, dt, _hoodie_commit_time
FROM orders
ORDER BY order_id
""").show(truncate=False)

spark.stop()
PY

提交脚本：

spark-submit \
  --master local[2] \
  --jars /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar \
  /tmp/hudi_spark_datasource_demo.py

执行后 Snapshot Query 结果如下：

=== snapshot query ===
+--------+-------+------+-------+----------+-------------------+
|order_id|user_id|amount|status |dt        |_hoodie_commit_time|
+--------+-------+------+-------+----------+-------------------+
|7001    |u101   |99.8  |CREATED|2025-08-04|20250805163423130  |
|7002    |u102   |168.0 |PAID   |2025-08-04|20250805163437087  |
|7003    |u103   |39.9  |CREATED|2025-08-05|20250805163423130  |
|7004    |u104   |258.6 |CREATED|2025-08-05|20250805163437087  |
+--------+-------+------+-------+----------+-------------------+

可以看到：

  1. 7001 和 7003 来自第一次 bulk_insert。
  2. 7002 被第二次 upsert 更新为 PAID。
  3. 7004 是第二次 upsert 插入的新记录。
  4. _hoodie_commit_time 能看出不同记录来自哪个提交。

2.1.1 DataSource 删除和覆盖

DataSource 方式除了 bulk_insert、insert、upsert，还经常用到 delete 和 insert_overwrite。

删除时不要直接删 HDFS 上的 Parquet 文件，而是让 Hudi 生成新的 commit。删除记录至少要带上 record key 和 partition path，Hudi 才能定位到对应 FileGroup：

deletes = spark.createDataFrame([
    (7001, "u101", 99.8, "CREATED", 1754296860000, "2025-08-04")
], columns)

(deletes.write.format("hudi")
    .options(**common_options)
    .option("hoodie.datasource.write.operation", "delete")
    .mode("append")
    .save(base_path))

覆盖适合分区重算。例如 2025-08-05 这个分区的数据来自上游离线重跑结果，与其逐条 upsert，不如直接让 Hudi 对输入数据涉及的分区执行 insert_overwrite：

recomputed = spark.createDataFrame([
    (7003, "u103", 45.0, "REPAIRED", 1754382120000, "2025-08-05"),
    (7005, "u105", 88.0, "CREATED", 1754382180000, "2025-08-05")
], columns)

(recomputed.write.format("hudi")
    .options(**common_options)
    .option("hoodie.datasource.write.operation", "insert_overwrite")
    .mode("append")
    .save(base_path))

这里有一个容易踩坑的点：第一次建表可以用 mode("overwrite") 清理旧路径，但后续正常写入、删除、覆盖分区时一般都用 mode("append")。真正的 Hudi 写入语义由 hoodie.datasource.write.operation 决定，不要把 Spark 的 SaveMode 和 Hudi 的 operation 混为一谈。

2.2 增量查询和时间旅行

前面几篇已经讲过 Snapshot Query、Incremental Query、Read Optimized Query，也提到过 Time Travel。本文示例是 COW 表，Read Optimized 主要针对 MOR 表，所以这里重点补充 Spark DataSource 中的 Incremental Query 和 Time Travel 读法。

从 Spark 视角看，这些查询方式不是多套表，而是同一张 Hudi 表在不同 Timeline 视角下的读取方式。

Spark DataSource 读取 Hudi 的几种常用视角

这张图主要说明 Snapshot、Incremental、Time Travel 是同一张 Hudi 表在不同 Timeline 视角下的读取方式。

为了让这一节可以单独复制执行，单独准备一个查询脚本：

cat > /tmp/hudi_spark_query_demo.py <<'PY'
from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("hudi-spark-query-demo")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog")
    .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
    .config("spark.kryo.registrator", "org.apache.spark.HoodieSparkKryoRegistrar")
    .getOrCreate())

base_path = "/hudi/demo_spark_ds_orders"

spark.read.format("hudi").load(base_path).createOrReplaceTempView("orders")

print("=== commits ===")
commits = [
    r[0]
    for r in spark.sql(
        "SELECT DISTINCT _hoodie_commit_time FROM orders ORDER BY _hoodie_commit_time"
    ).collect()
]
for commit in commits:
    print(commit)

print("=== incremental after first commit ===")
inc = (spark.read.format("hudi")
    .option("hoodie.datasource.query.type", "incremental")
    .option("hoodie.datasource.read.begin.instanttime", commits[0])
    .load(base_path))

inc.select("order_id", "status", "dt", "_hoodie_commit_time") \
    .orderBy("order_id") \
    .show(truncate=False)

print("=== time travel at first commit ===")
(spark.read.format("hudi")
    .option("as.of.instant", commits[0])
    .load(base_path)
    .select("order_id", "status", "dt", "_hoodie_commit_time")
    .orderBy("order_id")
    .show(truncate=False))

spark.stop()
PY

执行命令：

spark-submit \
  --master local[2] \
  --jars /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar \
  /tmp/hudi_spark_query_demo.py

输出 commit 列表：

=== commits ===
20250805163423130
20250805163437087

增量查询输出如下：

输出如下：

=== incremental after first commit ===
+--------+-------+----------+-------------------+
|order_id|status |dt        |_hoodie_commit_time|
+--------+-------+----------+-------------------+
|7001    |CREATED|2025-08-04|20250805163423130  |
|7002    |PAID   |2025-08-04|20250805163437087  |
|7003    |CREATED|2025-08-05|20250805163423130  |
|7004    |CREATED|2025-08-05|20250805163437087  |
+--------+-------+----------+-------------------+

这里要特别注意：这个小例子里直接使用第一个 commit 作为 begin.instanttime，因此当前 Hudi 1.0.2 实测结果会包含第一个提交相关的记录；如果只想读取第二次提交之后的增量，应把 begin instant 设置为第一个提交之后的时间，或结合 end instant / checkpoint 语义做验证。Hudi 1.0.x 完成态 Instant 文件名包含 begin/completion time，增量读取参数语义应以当前版本文档和实测为准。

不要把 Incremental Query 简单理解成数据库 binlog 的每一条变更明细。真正 CDC 语义还要结合上游变更类型、payload、去重和消费 checkpoint。

本文使用 Spark DataSource 的 as.of.instant 参数进行时间旅行查询；Spark SQL 里也可以使用 TIMESTAMP AS OF 等语法，具体支持和时间格式以 Hudi 1.0.2 文档和当前 Spark SQL 扩展为准。

输出如下：

=== time travel at first commit ===
+--------+-------+----------+-------------------+
|order_id|status |dt        |_hoodie_commit_time|
+--------+-------+----------+-------------------+
|7001    |CREATED|2025-08-04|20250805163423130  |
|7002    |CREATED|2025-08-04|20250805163423130  |
|7003    |CREATED|2025-08-05|20250805163423130  |
+--------+-------+----------+-------------------+

时间旅行能看到第一个 commit 时的表状态：7002 还没有更新成 PAID，7004 也还没有出现。

2.3 查看 HDFS 文件和 Timeline

查看表目录：

hdfs dfs -ls -R /hudi/demo_spark_ds_orders | sed -n '1,80p'

部分输出如下：

drwxr-xr-x   - root supergroup          0 2025-08-05 16:34 /hudi/demo_spark_ds_orders/.hoodie
-rw-r--r--   1 root supergroup       1278 2025-08-05 16:34 /hudi/demo_spark_ds_orders/.hoodie/hoodie.properties
drwxr-xr-x   - root supergroup          0 2025-08-05 16:34 /hudi/demo_spark_ds_orders/.hoodie/timeline
-rw-r--r--   1 root supergroup          0 2025-08-05 16:34 /hudi/demo_spark_ds_orders/.hoodie/timeline/20250805163423130.commit.requested
-rw-r--r--   1 root supergroup          0 2025-08-05 16:34 /hudi/demo_spark_ds_orders/.hoodie/timeline/20250805163423130.inflight
-rw-r--r--   1 root supergroup       4668 2025-08-05 16:34 /hudi/demo_spark_ds_orders/.hoodie/timeline/20250805163423130_20250805163435902.commit
-rw-r--r--   1 root supergroup          0 2025-08-05 16:34 /hudi/demo_spark_ds_orders/.hoodie/timeline/20250805163437087.commit.requested
-rw-r--r--   1 root supergroup       3600 2025-08-05 16:34 /hudi/demo_spark_ds_orders/.hoodie/timeline/20250805163437087.inflight
-rw-r--r--   1 root supergroup       4426 2025-08-05 16:34 /hudi/demo_spark_ds_orders/.hoodie/timeline/20250805163437087_20250805163440689.commit
drwxr-xr-x   - root supergroup          0 2025-08-05 16:34 /hudi/demo_spark_ds_orders/2025-08-04
-rw-r--r--   1 root supergroup     435620 2025-08-05 16:34 /hudi/demo_spark_ds_orders/2025-08-04/392aae5b-dfcb-47fa-8700-8ed91facabf2-0_0-42-75_20250805163437087.parquet
-rw-r--r--   1 root supergroup     435563 2025-08-05 16:34 /hudi/demo_spark_ds_orders/2025-08-04/392aae5b-dfcb-47fa-8700-8ed91facabf2-0_1-16-0_20250805163423130.parquet
drwxr-xr-x   - root supergroup          0 2025-08-05 16:34 /hudi/demo_spark_ds_orders/2025-08-05
-rw-r--r--   1 root supergroup     435789 2025-08-05 16:34 /hudi/demo_spark_ds_orders/2025-08-05/392aae5b-dfcb-47fa-8700-8ed91facabf2-1_1-42-76_20250805163437087.parquet

这里能看到两个核心事实：

  1. Spark 写入后，Hudi 在 .hoodie/timeline 下生成 commit 元数据。
  2. COW 表的业务数据以 Parquet Base File 形式存在于分区目录中。

在 Hudi 1.0.x 中，完成态 Instant 文件名里可以看到 begin instant 和 completion instant，例如 begin_completion.commit；这里前半段是请求 / 开始时间，后半段是完成时间。

本文 DataSource 示例没有开启 hoodie.datasource.write.hive_style_partitioning=true，因此分区目录直接是 2025-08-04、2025-08-05，而不是 dt=2025-08-04、dt=2025-08-05。

2.4 Spark SQL 操作 Hudi

Spark SQL 更适合交互式验证和数据修复。先准备 SQL 文件：

SQL 写法看起来更接近数据库，但底层仍然会经过 Hudi Catalog、Hudi Spark SQL Extension 和 Hudi Writer。

Spark SQL 操作 Hudi 表

这张图主要说明 Spark SQL 的 UPDATE、MERGE、DELETE 并不是直接改 Parquet 文件，而是通过 Hudi Catalog、SQL Extension 和 Writer 生成新的 Timeline Instant。

cat > /tmp/hudi_spark_sql_demo.sql <<'SQL'
SET spark.sql.shuffle.partitions=2;
SET hoodie.spark.sql.insert.into.operation=bulk_insert;

DROP TABLE IF EXISTS demo_spark_sql_orders;
DROP TABLE IF EXISTS demo_spark_sql_adjustment;

CREATE TABLE demo_spark_sql_orders (
  order_id BIGINT,
  user_id STRING,
  amount DOUBLE,
  status STRING,
  ts BIGINT,
  dt STRING
) USING HUDI
PARTITIONED BY (dt)
TBLPROPERTIES (
  'primaryKey' = 'order_id',
  'preCombineField' = 'ts',
  'type' = 'cow'
)
LOCATION '/hudi/demo_spark_sql_orders';

INSERT INTO demo_spark_sql_orders VALUES
  (7101, 'u201', 88.8, 'CREATED', 1754304000000, '2025-08-04'),
  (7102, 'u202', 176.5, 'CREATED', 1754304060000, '2025-08-04'),
  (7103, 'u203', 67.0, 'CREATED', 1754382600000, '2025-08-05');

UPDATE demo_spark_sql_orders
SET status = 'PAID', ts = 1754304120000
WHERE order_id = 7102;

CREATE TABLE demo_spark_sql_adjustment (
  order_id BIGINT,
  user_id STRING,
  amount DOUBLE,
  status STRING,
  ts BIGINT,
  dt STRING
) USING HUDI
TBLPROPERTIES (
  'primaryKey' = 'order_id',
  'preCombineField' = 'ts',
  'type' = 'cow'
)
LOCATION '/hudi/demo_spark_sql_adjustment';

INSERT INTO demo_spark_sql_adjustment VALUES
  (7103, 'u203', 70.0, 'PAID', 1754382660000, '2025-08-05'),
  (7104, 'u204', 199.0, 'CREATED', 1754382720000, '2025-08-05');

MERGE INTO demo_spark_sql_orders AS target
USING demo_spark_sql_adjustment AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

DELETE FROM demo_spark_sql_orders WHERE order_id = 7101;

SELECT order_id, user_id, amount, status, dt, _hoodie_commit_time
FROM demo_spark_sql_orders
ORDER BY order_id;
SQL

这里设置 hoodie.spark.sql.insert.into.operation=bulk_insert 主要用于前面的初始化 INSERT INTO；后续 UPDATE、MERGE、DELETE 会走各自的 Hudi SQL DML 路径。生产环境中不要长期固定所有 INSERT 都用 bulk_insert，应该按写入语义选择。

UPDATE SET * 和 INSERT * 只适合 source 与 target 字段完全一致的示例。生产环境更建议显式列出字段映射，避免 schema 变化或字段顺序导致误写。

执行 spark-sql：

hdfs dfs -rm -r -f /hudi/demo_spark_sql_orders /hudi/demo_spark_sql_adjustment

spark-sql \
  --master local[2] \
  --jars /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalogImplementation=in-memory \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar \
  -f /tmp/hudi_spark_sql_demo.sql

因为本文重点是验证 Hudi SQL DML，而当前测试环境的 Spark 3.5 与 Hive 4 Metastore 存在兼容问题，所以这里使用 in-memory catalog，避免连接 HMS；同时通过 HoodieCatalog 和 HoodieSparkSessionExtension 让 Spark SQL 识别 Hudi 表和 DML。执行前先 hdfs dfs -rm 是为了保证本地测试可重复运行，真实环境不要手工删除 Hudi 表目录；应通过 Hudi 表语义和治理流程处理清理、回滚或重建。

最终查询结果如下：

7102	u202	176.5	PAID	2025-08-04	20250805163609552
7103	u203	70.0	PAID	2025-08-05	20250805163619049
7104	u204	199.0	CREATED	2025-08-05	20250805163619049
Time taken: 0.544 seconds, Fetched 3 row(s)

这个结果说明：

  1. 7101 被 DELETE 删除。
  2. 7102 被 UPDATE 更新为 PAID。
  3. 7103 被 MERGE INTO 更新金额和状态。
  4. 7104 被 MERGE INTO 插入。

再看 Timeline：

hdfs dfs -ls /hudi/demo_spark_sql_orders/.hoodie/timeline | tail -20

输出如下：

-rw-r--r--   1 root supergroup          0 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163553272.commit.requested
-rw-r--r--   1 root supergroup          0 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163553272.inflight
-rw-r--r--   1 root supergroup       4694 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163553272_20250805163605660.commit
-rw-r--r--   1 root supergroup          0 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163609552.commit.requested
-rw-r--r--   1 root supergroup       3545 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163609552.inflight
-rw-r--r--   1 root supergroup       4186 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163609552_20250805163612401.commit
-rw-r--r--   1 root supergroup          0 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163619049.commit.requested
-rw-r--r--   1 root supergroup       3537 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163619049.inflight
-rw-r--r--   1 root supergroup       4172 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163619049_20250805163622388.commit
-rw-r--r--   1 root supergroup          0 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163623421.commit.requested
-rw-r--r--   1 root supergroup       3545 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163623421.inflight
-rw-r--r--   1 root supergroup       4176 2025-08-05 16:36 /hudi/demo_spark_sql_orders/.hoodie/timeline/20250805163623421_20250805163624997.commit

Spark SQL 每次写入也会生成 Timeline Instant。也就是说，Spark SQL 不是绕过 Hudi 直接改 Parquet，它仍然走 Hudi 表管理逻辑。

2.4.1 Spark SQL 其他常用操作

上面的 SQL 文件把 CREATE TABLE、INSERT、UPDATE、MERGE INTO、DELETE 跑通了。实际使用 Spark SQL 管 Hudi 表时，还会遇到下面几类操作。

创建 MOR 表：

CREATE TABLE demo_spark_sql_mor_orders (
  order_id BIGINT,
  user_id STRING,
  amount DOUBLE,
  status STRING,
  ts BIGINT,
  dt STRING
) USING HUDI
PARTITIONED BY (dt)
TBLPROPERTIES (
  'primaryKey' = 'order_id',
  'preCombineField' = 'ts',
  'type' = 'mor'
)
LOCATION '/hudi/demo_spark_sql_mor_orders';

把已有 Hudi 路径注册成 Spark SQL 表：

CREATE TABLE demo_spark_sql_orders_existing
USING HUDI
PARTITIONED BY (dt)
LOCATION '/hudi/demo_spark_sql_orders';

这种方式不会重新写数据，适合把已经存在的 Hudi 表挂到当前 Spark Catalog 下。如果已有路径下是完整 Hudi 表，Hudi 可以从 .hoodie/hoodie.properties 读取表配置；如果当前版本无法自动识别，需要补充 primaryKey、preCombineField、type 等 TBLPROPERTIES，或通过 Hive Sync / Metastore 统一注册。

CTAS 建表：

CREATE TABLE demo_spark_sql_ctas_orders
USING HUDI
PARTITIONED BY (dt)
TBLPROPERTIES (
  'primaryKey' = 'order_id',
  'preCombineField' = 'ts',
  'type' = 'cow'
)
LOCATION '/hudi/demo_spark_sql_ctas_orders'
AS
SELECT
  7301L AS order_id,
  'u401' AS user_id,
  66.6D AS amount,
  'CREATED' AS status,
  1754305200000L AS ts,
  '2025-08-04' AS dt;

CTAS 通常用于初始化小表或从临时表快速生成 Hudi 表。大批量初始化还是建议明确设置并行度、文件大小和 bulk_insert 相关参数。

INSERT OVERWRITE 修复分区：

INSERT OVERWRITE TABLE demo_spark_sql_orders PARTITION (dt = '2025-08-05')
SELECT
  7199L AS order_id,
  'u299' AS user_id,
  33.3D AS amount,
  'REPAIRED' AS status,
  1754383200000L AS ts;

这个语义适合修复某个分区，不适合做普通更新。普通更新应该用 UPDATE、MERGE INTO 或 DataSource upsert。

修改表结构和表属性：

ALTER TABLE demo_spark_sql_orders ADD COLUMNS (remark STRING);

ALTER TABLE demo_spark_sql_orders SET TBLPROPERTIES (
  'hoodie.keep.max.commits' = '20'
);

Schema Evolution 能解决不少字段新增问题，但字段类型变更、删除字段、修改 Hudi record key 这类操作要谨慎验证，不能只靠一条 ALTER TABLE 命令拍脑袋执行。

删除分区：

ALTER TABLE demo_spark_sql_orders DROP PARTITION (dt = '2025-08-05');

删除分区会影响这个分区的查询可见性，执行前最好先确认是否有下游增量消费还依赖相关 commit。DROP PARTITION 是否可用、是否转换为 Hudi 表语义，需要以 Hudi 1.0.2 Spark SQL 实测和当前 Catalog 配置为准。生产环境更建议先在测试表验证，不要直接对重要分区执行。

查看提交记录：

CALL show_commits(table => 'demo_spark_sql_orders', limit => 10);

Hudi Spark SQL procedures 很适合做运维排查，比如看 commits、files、clean、compaction 等元信息。不同 Hudi 小版本 procedure 名称和参数可能会调整，使用前可以先查当前版本文档或在测试环境里验证。如果 procedure 不可用，可以先退回到 hdfs dfs -ls /hudi/demo_spark_sql_orders/.hoodie/timeline 排查 Timeline。

2.5 IDEA 或 Maven 程序怎么写

生产里更常见的是把逻辑写成 Spark 应用，然后用 spark-submit 提交。Maven 依赖可以参考下面写法：

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>3.5.6</version>
        <scope>provided</scope>
    </dependency>

    <dependency>
        <groupId>org.apache.hudi</groupId>
        <artifactId>hudi-spark3.5-bundle_2.12</artifactId>
        <version>1.0.2</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

这里把 Spark 和 Hudi 都设置为 provided，是因为运行时由集群 Spark 环境和 spark-submit --jars 提供依赖；如果要打 fat jar，需要重新设计 shade 策略，避免和集群自带 Spark / Hadoop / Hudi 依赖冲突。

工程里我一般会按操作拆成几个小入口，方便单独提交和排查：

程序 主要逻辑 对应 Hudi operation
InsertDemo 初始化或纯追加写入 bulk_insert / insert
QueryDemo Snapshot、Incremental、Time Travel 查询 read options
UpdateDemo 按 record key 更新数据 upsert
DeleteDemo 按 record key 和 partition path 删除 delete
InsertOverwriteDemo 分区重算或表重算 insert_overwrite / insert_overwrite_table

这样拆的好处是参数边界更清楚。比如 DeleteDemo 必须确认输入里有 Hudi record key 和 partition path 字段，InsertOverwriteDemo 必须确认覆盖范围，不能和普通 upsert 任务混在一起。

一个最小 Java 示例：

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.util.HashMap;
import java.util.Map;

public class HudiSparkInsertDemo {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("HudiSparkInsertDemo")
                .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.hudi.catalog.HoodieCatalog")
                .config("spark.sql.extensions", "org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
                .config("spark.kryo.registrator", "org.apache.spark.HoodieSparkKryoRegistrar")
                .getOrCreate();

        Dataset<Row> df = spark.sql(
                "SELECT 7201L AS order_id, 'u301' AS user_id, 129.9D AS amount, " +
                "'CREATED' AS status, 1754304600000L AS ts, '2025-08-04' AS dt"
        );

        Map<String, String> options = new HashMap<>();
        options.put("hoodie.table.name", "demo_java_hudi_orders");
        options.put("hoodie.datasource.write.recordkey.field", "order_id");
        options.put("hoodie.datasource.write.partitionpath.field", "dt");
        options.put("hoodie.datasource.write.precombine.field", "ts");
        options.put("hoodie.datasource.write.table.type", "COPY_ON_WRITE");
        options.put("hoodie.datasource.write.keygenerator.class", "org.apache.hudi.keygen.SimpleKeyGenerator");
        options.put("hoodie.datasource.write.operation", "bulk_insert");

        df.write()
                .format("hudi")
                .options(options)
                .mode(SaveMode.Overwrite)
                .save("/hudi/demo_java_hudi_orders");

        spark.stop();
    }
}

提交时仍然要带 Hudi bundle：

spark-submit \
  --master yarn \
  --deploy-mode client \
  --class com.example.HudiSparkInsertDemo \
  --jars /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar \
hudi-spark-demo-1.0-SNAPSHOT.jar

这里演示的是生产 / 准生产提交方式，所以使用 --master yarn；如果只是沿用本文单机 local 测试，也可以改成 --master local[2]。YARN 模式还需要确认 HDFS、YARN、依赖分发和日志路径都已经配置好。

这里 spark-sql、spark-shell、spark-submit 的参数最好保持一致。否则很容易出现 shell 能跑、提交任务不能跑，或者 DataSource 能写、SQL 不能识别 Hudi 语法的问题。

3. 常见问题与实践判断

3.1 常见问题

3.1.1 ClassNotFoundException: HoodieCatalog

现象类似：

java.lang.ClassNotFoundException: org.apache.spark.sql.hudi.catalog.HoodieCatalog

常见原因是没有加载 Hudi Spark bundle，或者 bundle 版本与 Spark 版本不匹配。

修复方式：

spark-shell \
  --jars /opt/module/hudi-1.0.2/lib/hudi-spark3.5-bundle_2.12-1.0.2.jar \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog \
  --conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension \
  --conf spark.kryo.registrator=org.apache.spark.HoodieSparkKryoRegistrar

验证方式是重新启动后执行 spark.read.format("hudi") 或 CREATE TABLE ... USING HUDI。

3.1.2 NoSuchMethodError 或 NoClassDefFoundError

这类错误多数是依赖冲突，尤其是 Spark、Hudi、Hadoop、Hive、Avro、Parquet 版本之间不匹配。

排查顺序：

  1. 先确认 Spark 版本。
  2. 再确认 Hudi bundle 是否是对应 Spark 版本。
  3. 尽量使用 Hudi 官方 bundle，不要把 Hudi 子模块 jar 零散放进 classpath。
  4. 如果放到了 $SPARK_HOME/jars，就不要再额外加载另一个不同版本的 Hudi bundle。

3.1.3 Spark SQL 连接 Hive Metastore 报 Invalid method name: get_table

在本文这套 Spark 3.5.6、Hive 4.0.1、Hudi 1.0.2 的单机测试环境中，我真实遇到了这个问题。第一次直接跑 spark-sql 时，环境读取了 Hive 4 Metastore 配置，执行 DROP TABLE 阶段报错：

org.apache.hadoop.hive.ql.metadata.HiveException:
Unable to fetch table demo_spark_sql_orders. Invalid method name: 'get_table'

原因是 Spark 3.5 内置 Hive client 与当前 Hive 4 Metastore 服务存在 Thrift 接口兼容问题。本文重点是验证 Spark SQL 操作 Hudi，所以最终使用本地 catalog 避开 HMS：

--conf spark.sql.catalogImplementation=in-memory

如果生产环境必须通过 Hive Metastore 管理表，建议单独验证 Spark、Hive、Hudi、HMS client 的兼容组合，不要只看单个组件版本是否够新。

3.1.4 Hudi SQL 语法不生效

如果 UPDATE、MERGE INTO 或 DELETE 不能识别，优先检查：

--conf spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension
--conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog

没有这两个配置时，Spark SQL 会按普通 Spark 表理解语句，自然识别不了 Hudi 扩展能力。

3.1.5 增量查询不是完整 binlog

Hudi Incremental Query 适合做增量 pipeline，但它不是 MySQL binlog 原样回放。它返回的是 Hudi commit 范围内的增量结果视图。是否能表达 I/U/D 变更语义，还要看上游 CDC 数据、写入方式和 payload 设计。后面讲 Flink CDC 入湖时还会展开这个问题。

如果需要更接近数据库 binlog 的 before / after 或操作类型语义，需要单独验证 Hudi CDC Query / changelog 相关能力，以及写入时是否保留 CDC 信息。

3.2 Spark 集成实践判断

Spark 集成 Hudi 的核心不是“让 Spark 能读写一个新格式”，而是让 Spark 能参与 Hudi 的表管理。

DataSource 更像工程接口，适合写在批任务里；Spark SQL 更适合交互式建表、调试、临时修数，也可以封装成生产 SQL 作业；只是它对 Catalog、Extension、Metastore 配置更敏感。spark-submit 则是把前两者变成生产任务的提交方式。三者的底层差异没有想象中大，真正重要的是 Hudi 表参数、Timeline、索引和文件布局。

我更建议先用 Spark DataSource 把基本写入跑通，再用 Spark SQL 验证 DML。原因很简单：DataSource 参数更显式，容易看清楚 record key、partition path、precombine 和 operation；SQL 写起来方便，但如果 Catalog 或 Hive Metastore 配置混乱，排错成本反而更高。

另外，Spark 很适合 Hudi 的离线初始化和分区修复，但不一定适合所有实时写入。实时 CDC 写入更适合放到 Flink 里，这也是后面几篇要继续讲 Flink Hudi 的原因。

4. 总结

这一篇主要完成了 Spark 集成 Hudi 的实战验证：

  1. 在本文 2025-08-05 的测试环境中，Hudi 1.0.2 与 Spark 3.5.6 是验证通过的组合。
  2. spark-shell、spark-sql、spark-submit 都需要加载 Hudi Spark bundle。
  3. DataSource 写入时要明确 record key、partition path、precombine field 和 operation，删除与覆盖也要通过 Hudi commit 完成。
  4. Spark SQL 可以操作 Hudi 表的 CREATE、INSERT、UPDATE、MERGE INTO、DELETE、INSERT OVERWRITE、ALTER TABLE 和 procedures。
  5. Snapshot Query、Incremental Query、Time Travel 都可以通过 Spark DataSource 查询。
  6. Hive Metastore 不是越新越好，Spark SQL 与 Hive 4 HMS 的兼容性需要单独验证。

到这里，Spark 这条线已经可以完成 Hudi 表的基础读写、查询和修复。下一篇会继续看 DeltaStreamer，重点是 Kafka 数据如何通过 Hudi 官方摄入工具进入数据湖。

文章作者: hnbian
文章链接: https://www.hnbian.cn/posts/7f3a9c21.html
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !
spark hudi data-lake lakehouse spark-sql
评论
  目录