Apache Hudi 学习笔记 14：Hudi 底层读写原理解析

hudi

发布日期: 2025-10-19

文章字数: 6.9k

阅读时长: 29 分

阅读次数:

前面已经从表类型、索引、Spark 集成、Flink 集成、CDC、调优和高级能力几个角度把 Hudi 跑了一遍。到这篇，我想把视角往下一层放：同样是写入一条订单更新，Hudi 内部到底怎样判断新旧、怎样找到文件、怎样把 Log File 和 Base File 合成最新结果、怎样清理旧版本。

这篇不会只停留在抽象图。为了让原理能落到真实文件上，我在服务器上重新写了一张 MOR 表，连续对同一个 order_id 做了多次 upsert，然后把 HDFS 目录、Timeline、查询结果、Cleaner 日志都复制出来。这样看 FileGroup、FileSlice、Log File、preCombine 和 Clean 时，心里会更有底。

1. 验证表和写入机制

1.1 背景问题

Hudi 的很多能力表面看是参数问题，本质是底层机制问题。

比如：

同一个 record key / 业务主键一批数据里出现多次，为什么最终只保留一条。
COW 表为什么更新慢，但查询简单。
MOR 表为什么更新快，但查询时要合并 Log File。
Timeline 里为什么有 .requested、.inflight、最终完成文件。
Compaction 为什么不是简单把小文件合并。
Clean 为什么不能随便调得太激进。
Flink 写 Hudi 为什么强依赖 checkpoint。

如果只记参数，很容易碰到现象就乱调。理解读写路径以后，很多问题会变成比较明确的判断：是 key 生成问题、索引定位问题、写入合并问题、表服务问题，还是查询视图问题。

1.2 本文使用的验证表

这篇用 Spark DataSource 创建一张 MOR 表，路径是：

hdfs:///hudi/demo_hudi_internal_orders

因为当前 Spark 环境默认会尝试连接 Hive Metastore，而 Spark 3.5.6 与 Hive 4.0.1 的 Thrift 方法存在兼容问题，所以本文的验证代码使用 spark.sql.catalogImplementation=in-memory，避免把重点放到 Hive 元数据上。

启动命令如下：

$SPARK_HOME/bin/spark-shell \
  --master local[2] \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  --conf spark.sql.catalogImplementation=in-memory \
  -i /tmp/hudi14_spark.scala

核心写入代码如下：

import spark.implicits._

val basePath = "hdfs:///hudi/demo_hudi_internal_orders"

val commonOptions = Map(
  "hoodie.table.name" -> "demo_hudi_internal_orders",
  "hoodie.datasource.write.table.type" -> "MERGE_ON_READ",
  "hoodie.datasource.write.operation" -> "upsert",
  "hoodie.datasource.write.recordkey.field" -> "order_id",
  "hoodie.datasource.write.partitionpath.field" -> "dt",
  "hoodie.datasource.write.precombine.field" -> "ts",
  "hoodie.metadata.enable" -> "false",
  "hoodie.compact.inline" -> "false",
  "hoodie.clean.automatic" -> "true",
  "hoodie.clean.policy" -> "KEEP_LATEST_COMMITS",
  "hoodie.clean.commits.retained" -> "2"
)

val batch1 = Seq(
  (15001L, "u501", BigDecimal("99.90"), "CREATED", 1760783600000L, "2025-10-18"),
  (15002L, "u502", BigDecimal("168.00"), "CREATED", 1760783660000L, "2025-10-18")
).toDF("order_id", "user_id", "amount", "status", "ts", "dt")

batch1.write.format("hudi").mode("overwrite")
  .option("hoodie.table.name", "demo_hudi_internal_orders")
  .option("hoodie.datasource.write.table.type", "MERGE_ON_READ")
  .option("hoodie.datasource.write.operation", "upsert")
  .option("hoodie.datasource.write.recordkey.field", "order_id")
  .option("hoodie.datasource.write.partitionpath.field", "dt")
  .option("hoodie.datasource.write.precombine.field", "ts")
  .option("hoodie.metadata.enable", "false")
  .option("hoodie.compact.inline", "false")
  .option("hoodie.clean.automatic", "true")
  .option("hoodie.clean.policy", "KEEP_LATEST_COMMITS")
  .option("hoodie.clean.commits.retained", "2")
  .save(basePath)

第二批写入对 15001 做更新，并新增 15003：

val batch2 = Seq(
  (15001L, "u501", BigDecimal("99.90"), "PAID", 1760783960000L, "2025-10-18"),
  (15003L, "u503", BigDecimal("49.00"), "CREATED", 1760784020000L, "2025-10-19")
).toDF("order_id", "user_id", "amount", "status", "ts", "dt")

batch2.write.format("hudi").mode("append")
  .option("hoodie.table.name", "demo_hudi_internal_orders")
  .option("hoodie.datasource.write.table.type", "MERGE_ON_READ")
  .option("hoodie.datasource.write.operation", "upsert")
  .option("hoodie.datasource.write.recordkey.field", "order_id")
  .option("hoodie.datasource.write.partitionpath.field", "dt")
  .option("hoodie.datasource.write.precombine.field", "ts")
  .option("hoodie.metadata.enable", "false")
  .option("hoodie.compact.inline", "false")
  .option("hoodie.clean.automatic", "true")
  .option("hoodie.clean.policy", "KEEP_LATEST_COMMITS")
  .option("hoodie.clean.commits.retained", "2")
  .save(basePath)

后面我又追加了两次 15001 的状态更新：

Seq(
  Seq((15001L, "u501", BigDecimal("99.90"), "SHIPPED", 1760784260000L, "2025-10-18")),
  Seq((15001L, "u501", BigDecimal("99.90"), "FINISHED", 1760784560000L, "2025-10-18"))
).foreach { rows =>
  rows.toDF("order_id", "user_id", "amount", "status", "ts", "dt")
    .write.format("hudi")
    .mode("append")
    .options(commonOptions)
    .save(basePath)
}

这里的 ts 是模拟业务事件时间，也作为 Hudi 的 precombine.field 排序字段使用；它不代表写入任务实际提交时间。真正的写入提交时间要看 _hoodie_commit_time 和 Timeline。

最终查询代码如下：

spark.read.format("hudi")
  .load(basePath)
  .select("order_id", "status", "dt", "_hoodie_commit_time", "_hoodie_file_name")
  .orderBy("order_id")
  .show(false)

最终查询结果如下：

+--------+--------+----------+-------------------+------------------------------------------------------------------------+
|order_id|status  |dt        |_hoodie_commit_time|_hoodie_file_name                                                       |
+--------+--------+----------+-------------------+------------------------------------------------------------------------+
|15001   |FINISHED|2025-10-18|20251019183240817  |948d3df1-03e5-4ef8-a626-123f6feb39e8-0                                  |
|15002   |CREATED |2025-10-18|20251019182929097  |948d3df1-03e5-4ef8-a626-123f6feb39e8-0_0-13-16_20251019182929097.parquet|
|15003   |CREATED |2025-10-19|20251019182935202  |4961d719-e756-43ff-8999-8b00c7a523f7-0_1-35-40_20251019182935202.parquet|
+--------+--------+----------+-------------------+------------------------------------------------------------------------+

注意 15001 的 _hoodie_file_name 没有显示成完整 Parquet 文件名，而是 FileId。结合本文这张 MOR 表的目录结构和同 FileGroup 下多个 Log File，可以推断它的最新状态很可能来自 Snapshot Query 对 Base File 与 Log File 的合并视图，而不是直接来自一个新的 Base File。

1.3 写入内部流程

Hudi 写入可以粗略拆成几步：

输入数据变成 Hudi 能识别的记录。
根据 record key 和 partition path 生成 HoodieKey。
同一个 key 在写入批次内先做 preCombine。
通过 Index 判断这条记录是 insert 还是 update。
找到对应 FileGroup。
COW 表重写 Base File，MOR 表追加 Log File。
写入 Timeline，完成事务提交。

Hudi 写入内部流程

这张图主要说明写入侧不是直接落 Parquet，而是经过 HoodieKey、preCombine、Index、FileGroup 定位和 Timeline 提交之后，才形成可被 Reader 识别的一致表状态。

用一句话概括：写入不是直接把数据写成 Parquet，而是先确定 key、确定文件位置、确定版本关系，再通过 Timeline 把这次写入变成一个可见的 instant。

1.3.1 HoodieKey：记录定位的起点

Hudi 不是按整行内容判断一条数据，而是按 record key + partition path 判断一条记录的位置。

本文写入配置里：

.option("hoodie.datasource.write.recordkey.field", "order_id")
.option("hoodie.datasource.write.partitionpath.field", "dt")

所以：

order_id=15001, dt=2025-10-18

会生成一条属于 2025-10-18 分区的记录。后续 Index 定位、FileGroup 选择、Log File 追加，都是围绕这个 key 展开的。

表属性中也能看到这些信息：

hdfs dfs -cat /hudi/demo_hudi_internal_orders/.hoodie/hoodie.properties \
  | egrep "hoodie.table.name|hoodie.table.type|hoodie.table.precombine.field|hoodie.table.recordkey.fields|hoodie.table.partition.fields|hoodie.table.version" \
  | sort

输出：

hoodie.table.name=demo_hudi_internal_orders
hoodie.table.partition.fields=dt
hoodie.table.precombine.field=ts
hoodie.table.recordkey.fields=order_id
hoodie.table.type=MERGE_ON_READ
hoodie.table.version=8

1.4 去重与合并：preCombine 不只是去重

Hudi 中经常说“去重”，但这个词容易误导。更准确地说，Hudi 是在写入前和写入时做“同 key 多版本合并”。

核心接口是 HoodieRecordPayload，里面有几个关键方法：

方法	作用
`preCombine`	同一批输入数据里，同一个 key 出现多次时，先合并出一条候选记录
`combineAndGetUpdateValue`	update 时，把新记录和存储中的旧记录合并，生成最终写回值
`getInsertValue`	insert 时，把新记录转成要写入存储的值

本文表使用：

.option("hoodie.datasource.write.precombine.field", "ts")

这表示同一个 order_id 出现多条记录时，ts 更大的那条更“新”。例如：

15001, CREATED,  ts=1760783600000
15001, PAID,     ts=1760783960000
15001, SHIPPED,  ts=1760784260000
15001, FINISHED, ts=1760784560000

最终查询结果是：

15001 FINISHED

这不是因为 Hudi 简单保留了最后一行文件，而是因为写入和读取过程中都围绕 record key、commit time、preCombine 字段来确定最新状态。

这里还有几个容易被忽略的细节：

如果同一个 key 的两条记录 precombine.field 相同，默认 payload 通常会以后到记录作为更新结果。
Flink 侧的 write.precombine 控制写入前 buffer 阶段是否先做同 key 合并；Spark DataSource 侧对应的是 hoodie.datasource.write.precombine.field 这类字段配置，两者不是同一个参数名。
即使写入前没有提前合并，COW 写入、MOR Snapshot 读取合并和 MOR compaction 阶段仍然会在 Base File + 增量数据 merge 时通过 combineAndGetUpdateValue 决定最终值。

也就是说，Hudi 的“去重”至少有两个层次：

阶段	位置	典型接口	作用
写入前 buffer 合并	writer 内存中	`HoodieRecordPayload#preCombine`	同一批输入里先挑出候选新记录
写入 / 压缩时 merge	Base File 与增量记录合并	`HoodieRecordPayload#combineAndGetUpdateValue`	新旧记录真正合并成存储结果

跨 partition 更新还要看索引策略。默认情况下，相同 record key 如果换了 partition path，旧分区里的记录不一定会自动消失；开启全局索引能力后，Hudi 才能先定位旧 partition 并发出删除，再把新记录写到新 partition。Flink State Index 如果按全局语义使用，需要确认 index.global.enabled 等参数在当前 Hudi 1.0.2 Flink connector 中的默认值和实际配置；Bucket Index 或其他索引策略下也要结合业务分区字段谨慎验证。

这里可以理解为：precombine.field 要选一个真正能表达业务新旧顺序的字段。CDC 场景里，如果只用普通更新时间，而这个字段可能乱序或回退，就可能把旧数据当成新数据写进去。

1.5 MOR 表写入：Base File 与 Log File 的关系

MOR 表的典型特点是：

insert 可以写 Base File。
update 在常见 upsert 路径下倾向追加 Log File。
查询最新快照时，Reader 合并 Base File 和 Log File。
Compaction 后，Log File 中的更新会合并进新的 Base File。

本文连续 upsert 以后，2025-10-18 分区目录如下：

hdfs dfs -ls /hudi/demo_hudi_internal_orders/2025-10-18

输出：

Found 5 items
-rw-r--r--   1 root supergroup       1294 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/2025-10-18/.948d3df1-03e5-4ef8-a626-123f6feb39e8-0_20251019182935202.log.1_0-35-39
-rw-r--r--   1 root supergroup       1297 2025-10-19 18:32 /hudi/demo_hudi_internal_orders/2025-10-18/.948d3df1-03e5-4ef8-a626-123f6feb39e8-0_20251019183233289.log.1_0-14-10
-rw-r--r--   1 root supergroup       1298 2025-10-19 18:32 /hudi/demo_hudi_internal_orders/2025-10-18/.948d3df1-03e5-4ef8-a626-123f6feb39e8-0_20251019183240817.log.1_0-39-30
-rw-r--r--   1 root supergroup         96 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/2025-10-18/.hoodie_partition_metadata
-rw-r--r--   1 root supergroup     436037 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/2025-10-18/948d3df1-03e5-4ef8-a626-123f6feb39e8-0_0-13-16_20251019182929097.parquet

这里能看到一个很典型的 FileGroup：

948d3df1-03e5-4ef8-a626-123f6feb39e8-0

它下面有：

一个 Base File：948d3df1-..._20251019182929097.parquet
三个 Log File：.948d3df1-..._20251019182935202.log.1_... 等

这就是 MOR 的核心：更新不一定立即重写 Parquet，在常见 upsert 路径下会先追加到同一个 FileGroup 的 Log File，后续由 Snapshot Query 读时合并，或由 Compaction 写出新的 Base File。

可以先用这个简化结构理解 FileGroup 和 FileSlice：

FileGroup: 948d3df1-...
  FileSlice@20251019182929097
    Base File: 948d3df1-..._20251019182929097.parquet
  FileSlice@20251019182935202
    Base File: 948d3df1-..._20251019182929097.parquet
    Log File:  .948d3df1-..._20251019182935202.log.1_...
  FileSlice@20251019183240817
    Base File: 948d3df1-..._20251019182929097.parquet
    Log Files: .948d3df1-..._20251019182935202.log.1_...
               .948d3df1-..._20251019183233289.log.1_...
               .948d3df1-..._20251019183240817.log.1_...

1.6 Timeline：写入状态不是文件名游戏

再看 Timeline：

hdfs dfs -ls /hudi/demo_hudi_internal_orders/.hoodie/timeline

输出：

Found 13 items
-rw-r--r--   1 root supergroup       3430 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019182929097.deltacommit.inflight
-rw-r--r--   1 root supergroup          0 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019182929097.deltacommit.requested
-rw-r--r--   1 root supergroup       4163 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019182929097_20251019182933809.deltacommit
-rw-r--r--   1 root supergroup       3600 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019182935202.deltacommit.inflight
-rw-r--r--   1 root supergroup          0 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019182935202.deltacommit.requested
-rw-r--r--   1 root supergroup       4462 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019182935202_20251019182939176.deltacommit
-rw-r--r--   1 root supergroup       3534 2025-10-19 18:32 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019183233289.deltacommit.inflight
-rw-r--r--   1 root supergroup          0 2025-10-19 18:32 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019183233289.deltacommit.requested
-rw-r--r--   1 root supergroup       4251 2025-10-19 18:32 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019183233289_20251019183239895.deltacommit
-rw-r--r--   1 root supergroup       3534 2025-10-19 18:32 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019183240817.deltacommit.inflight
-rw-r--r--   1 root supergroup          0 2025-10-19 18:32 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019183240817.deltacommit.requested
-rw-r--r--   1 root supergroup       4251 2025-10-19 18:32 /hudi/demo_hudi_internal_orders/.hoodie/timeline/20251019183240817_20251019183243048.deltacommit
drwxr-xr-x   - root supergroup          0 2025-10-19 18:29 /hudi/demo_hudi_internal_orders/.hoodie/timeline/history

Hudi 1.x 的活跃 Timeline 在 .hoodie/timeline 目录下，和很多旧资料中直接把 instant 文件放在 .hoodie 根目录的布局不同。排查时要以当前表的目录结构和 hoodie.properties 中的版本信息为准。

MOR 表里这里是 deltacommit，不是普通 commit。每次写入都有：

状态文件	含义
`.requested`	请求开始一次写入或表服务
`.inflight`	写入正在执行
最终完成文件	写入完成，对查询端可见

真正对读可见的是完成的 instant。Hudi Reader 不会简单扫描目录里所有 Parquet 和 Log File，而是根据 Timeline 选择一个一致的文件视图。

在 Hudi 1.x 的 Timeline 布局下，同一个 begin instant 对应的 .requested、.inflight 和完成态文件可能同时出现在目录里。判断是否已经提交，主要看完成态 instant；完成态文件名中的 begin_completion.deltacommit 也能看到 begin instant 和 completion instant。

这也是为什么 Hudi 能在文件系统上做出类似事务的效果：它不是让文件系统本身变成数据库，而是在文件系统之上维护了一层有序的提交日志。

1.7 Flink 写入原理：checkpoint 提交 instant

Spark 批写入通常是一个 job 提交一个或多个 instant。Flink 流式写入不一样，它要和 checkpoint 协调。

这一节是对前几篇 Flink 写入链路的原理回顾，不是本文这次 Spark DataSource MOR 实验的新增输出。

Flink Hudi 写入链路里，常见内部角色可以这样理解：

角色	作用
`RowDataToHoodieFunction`	把 Flink `RowData` 转成 Hudi 记录
`BucketAssigner`	给记录分配 bucket / FileGroup
`Hoodie Stream Writer`	持续写 Base File 或 Log File
`Operator Coordinator`	在 checkpoint 成功时协调提交 instant

这也是为什么 Flink 写 Hudi 必须认真配置 checkpoint。没有稳定 checkpoint，流式写入就很难形成稳定的 completed instant，下游读取任务也就无法获得一致的新快照。

更细一点看：

RowDataToHoodieFunction 先把 Flink 的 RowData 封装成 Hudi 能处理的记录。
BucketAssigner 判断写入位置：insert 倾向放入较小的 FileGroup，update 直接写到索引定位到的 current location。
Hoodie Stream Writer 把记录缓存在内存里，达到 flush 阈值或 checkpoint 触发时刷到文件系统。
Operator Coordinator 负责协调 checkpoint 事件，把这批写入提交成 Timeline 上的 completed instant，并为下一批写入分配新的 instant time。

所以 Flink Hudi 的写入吞吐和四类参数强相关：source 并行度、bucket assigner 并行度、writer buffer、checkpoint 间隔。只调 write.tasks，但 checkpoint 或 bucket assigner 卡住，吞吐并不会真正上去。

一个常见误区是：Flink SQL 提交后 HDFS 上没立刻看到数据，就以为写失败。实际要看：

job 是否 RUNNING。
checkpoint 是否完成。
.hoodie/timeline 是否出现 completed instant。
写入目录里是否只有临时 marker 或 inflight 文件。

前面几篇已经在 Flink SQL 里验证过这个问题，这里就不重复截图了。

2. 读取和表服务

2.1 读取内部流程

Hudi 读取也不是简单 ls 目录后读所有文件。它会先构建文件视图，再按查询类型读取。

Hudi 读取内部流程

这张图主要说明 Reader 会先读取 Timeline 和表元数据构造 FileSlice 视图，再按 Snapshot、Read Optimized、Incremental 或 Streaming 等读取视角选择不同文件集合。

不同查询方式的读路径大致如下：

查询方式	读取逻辑
Snapshot Query	读取最新快照；MOR 需要合并 Base File 和 Log File
Read Optimized Query	只读当前可用于读优化视图的 Base File；MOR 表中可能看不到尚未合并进 Base File 的最新 Log File 更新
Incremental Query	按 commit / deltacommit 范围读取变化
Streaming Query	持续监听 Timeline 中新增 instant

以本文这张 MOR 表为例，15001 的最新状态在 Log File 里，Snapshot Query 能读到 FINISHED。如果只读旧 Base File，就只能看到最早的 CREATED 状态。

所以 MOR 的查询延迟来自哪里就很清楚了：

需要读取 Base File。
需要扫描同 FileGroup 下的 Log File。
需要按 record key 合并多版本。
Log File 越多，合并成本越高。

这也是 Compaction 存在的原因。

2.2 Compaction：把读时合并变成写后合并

Compaction 主要解决 MOR 表的一个问题：Log File 积累太多以后，Snapshot Query 每次都要合并，读延迟会上升。

Compaction 做的事情是：

找到需要压缩的 FileSlice。
读取旧 Base File。
读取同 FileGroup 的 Log File。
合并出新的 Base File。
生成新的 FileSlice。
通过 Timeline 记录 compaction 状态。

Hudi Compaction 与 Clean

这张图主要说明 Compaction 负责把 MOR 的 Log File 合并回新的 Base File，Clean/Cleaner 则根据保留策略回收已经不再需要的旧 FileSlice。

Flink / Spark streaming writer 都可以异步调度 compaction，触发策略常见有四类：

`compaction.trigger.strategy`	含义
`num_commits`	delta commit 数达到阈值后触发
`time_elapsed`	距离上次 compaction 的时间超过阈值后触发
`num_and_time`	commit 数和时间条件同时满足后触发
`num_or_time`	commit 数或时间任一条件满足后触发

生产中常见做法是按 delta commit 数触发。这个策略不代表 compaction 一定马上完成，因为 compaction 本身消耗 CPU、内存和 IO，在线执行时还可能和写入链路抢资源。

2.2.1 本文 Compaction 未成功生成 plan 的边界情况

离线 Compaction 工具的尝试命令结构如下：

$SPARK_HOME/bin/spark-submit \
  --class org.apache.hudi.utilities.HoodieCompactor \
  --master local[2] \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  /opt/module/hudi-1.0.2/lib/hudi-utilities-bundle_2.12-1.0.2.jar \
  --base-path hdfs:///hudi/demo_hudi_internal_orders \
  --table-name demo_hudi_internal_orders \
  --mode scheduleAndExecute \
  --parallelism 1 \
  --strategy org.apache.hudi.table.action.compact.strategy.UnBoundedCompactionStrategy \
  --hoodie-conf hoodie.metadata.enable=false \
  --hoodie-conf hoodie.compact.inline=false \
  --spark-master local[2]

这次小表实际执行时，没有成功调度出 compaction plan：

INFO HoodieCompactor: Running Mode: [scheduleandexecute]
INFO HoodieCompactor: Step 1: Do schedule
WARN HoodieCompactor: No instant time is provided for scheduling compaction.
INFO BaseHoodieTableServiceClient: Scheduling table service COMPACT for table hdfs:///hudi/demo_hudi_internal_orders
INFO BaseHoodieTableServiceClient: Scheduling compaction at instant time: 20251019183308612 for table hdfs:///hudi/demo_hudi_internal_orders
INFO HoodieCompactionPlanGenerator: Compaction Strategy used is: org.apache.hudi.table.action.compact.strategy.UnBoundedCompactionStrategy@5ad40184
WARN HoodieCompactor: Couldn't do schedule
Exception in thread "main" org.apache.hudi.exception.HoodieException: Fail to run compaction for demo_hudi_internal_orders, return code: -1

这个结果我没有改写成“成功示例”，因为它本身就是一个值得记下来的边界：Compaction 不是看到 Log File 就一定执行成功。实际是否生成 plan，会受表状态、策略、instant、写入服务状态、配置和版本行为影响。生产里排查 Compaction 时，不能只看有没有 .log 文件，还要看 compaction instant、plan 生成日志和具体策略。

因此本文没有展示 compaction 成功后的新 Base File。这个命令更适合用来说明 HoodieCompactor 的调用结构和失败边界；真实环境必须按 Hudi 1.0.2 utilities bundle、表状态和当前参数重新验证。

2.2.2 Compaction 配置与流读影响

常见配置如下：

hoodie.compact.inline=false
hoodie.compact.inline.max.delta.commits=5
hoodie.compaction.strategy=org.apache.hudi.table.action.compact.strategy.LogFileSizeBasedCompactionStrategy

Flink SQL 常见配置是：

'compaction.schedule.enabled' = 'true',
'compaction.async.enabled' = 'true',
'compaction.trigger.strategy' = 'num_commits',
'compaction.delta_commits' = '5'

如果这张 MOR 表作为后续流式聚合的输入表，还要特别注意 read.streaming.skip_compaction。当下游已经消费过压缩前的 Log File，再去消费 compaction 生成的新 Base File，聚合类任务可能出现重复累加。这类流读聚合链路需要评估是否配置：

'read.streaming.skip_compaction' = 'true'

这个参数的含义不是“不让表执行 compaction”，而是 reader 跳过 compaction instant，避免把压缩产物再次当成新变更消费。

我的建议是：测试环境可以手动触发 Compaction 熟悉流程，生产环境更适合让写入任务或独立表服务按策略运行，并配合监控观察 Log File 数量、Snapshot Query 延迟和 compaction 任务耗时。

2.3 Clean：删除旧 FileSlice，不是删除旧数据

Clean 的职责是回收旧版本文件，避免表目录无限增长。

Hudi 写入链路中常见做法是在 commit 后 inline 触发 cleaning，常见策略是 KEEP_LATEST_COMMITS。不过是否自动触发、何时触发、保留多少旧版本，都要以当前写入配置和版本默认值为准。Clean 不能配置得过于激进，因为长查询、增量查询、流式读取都可能还需要旧 instant 对应的文件。

本文手动执行了一次 Cleaner：

$SPARK_HOME/bin/spark-submit \
  --class org.apache.hudi.utilities.HoodieCleaner \
  --master local[2] \
  --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \
  /opt/module/hudi-1.0.2/lib/hudi-utilities-bundle_2.12-1.0.2.jar \
  --target-base-path hdfs:///hudi/demo_hudi_internal_orders \
  --spark-master local[2] \
  --hoodie-conf hoodie.metadata.enable=false \
  --hoodie-conf hoodie.clean.policy=KEEP_LATEST_COMMITS \
  --hoodie-conf hoodie.clean.commits.retained=2

关键输出如下：

INFO HoodieCleaner: Creating Cleaner with configs : {hoodie.metadata.enable=false, hoodie.clean.policy=KEEP_LATEST_COMMITS, hoodie.clean.commits.retained=2}
INFO BaseHoodieTableServiceClient: Cleaner started for table hdfs:///hudi/demo_hudi_internal_orders
INFO BaseHoodieTableServiceClient: Scheduling cleaning at instant time: 20251019183349770 for table hdfs:///hudi/demo_hudi_internal_orders
INFO CleanPlanActionExecutor: Earliest commit to retain for clean : 20251019183233289
INFO CleanPlanActionExecutor: Total partitions to clean : 2, with policy KEEP_LATEST_COMMITS
INFO CleanPlanner: Cleaning 2025-10-18, retaining latest 2 commits.
INFO CleanPlanner: Cleaning 2025-10-19, retaining latest 2 commits.
INFO CleanPlanner: 0 patterns used to delete in partition path:2025-10-19
INFO CleanPlanner: 0 patterns used to delete in partition path:2025-10-18
INFO HoodieCleaner: Cleaner ran successfully

这里验证了 Cleaner 的计划和判断流程，但没有真正删除文件，因为这张小表没有足够多可安全回收的旧 FileSlice。日志里能看出 Clean 的判断方式：

先确定最早保留的 commit。
再扫描各分区。
按策略判断旧 FileSlice 是否可以删。
最后才真正删除匹配文件。

常见策略如下：

策略	含义
`KEEP_LATEST_COMMITS`	保留最近 N 个 commit，适合和增量查询窗口一起考虑
`KEEP_LATEST_FILE_VERSIONS`	每个 FileGroup 保留最近 N 个文件版本
`KEEP_LATEST_BY_HOURS`	按时间窗口保留旧版本

这里可以理解为：Clean 是存储成本和读一致性之间的平衡。clean.commits.retained 调小可以更快回收文件，但也会压缩增量查询和长查询的可回放窗口。

如果使用 KEEP_LATEST_FILE_VERSIONS，判断维度会从“最近多少个 commit”切到“每个 FileGroup 保留多少个文件版本”。这个策略不关心 commit 时间窗口，更适合希望每个 FileGroup 只保留固定版本数的场景。缺点是如果下游依赖较长的增量窗口，就要额外确认旧 instant 对应的文件没有被提前清掉。

2.4 流式读取：不是一直扫目录

Flink 流读 Hudi 时，经常会看到几个关键词：

split_monitor
split_reader
read.streaming.enabled
read.start-commit
clean.retain_commits

它的大致流程是：

Split Monitor 监听 Timeline。
发现新的 completed instant。
构建这个 instant 影响到的 FileSlice。
分发读取任务时尽量保持同一个 FileGroup 的 Base File / Log File 顺序关系一致，具体分发方式要以当前 Flink connector 实现为准。
Split Reader 读取 Base File / Log File 并输出结果。

这样做的原因是：同一个 FileGroup 下的 Base File 和 Log File 之间有顺序关系，不能被随意打散处理。Reader 侧要保证同一组文件能按一致的 FileSlice 视图合并。

所以流式读取最怕两类问题：

下游读得太慢，旧 commit 已经被 Clean 回收。
MOR 表 Log File 太多，读取合并速度跟不上写入速度。

这也是为什么第 12 篇里强调过：

'read.streaming.enabled' = 'true',
'read.start-commit' = '20250929203600',
'read.streaming.skip_compaction' = 'true'

以及 Clean 保留窗口要和消费延迟配套设置。

3. 常见问题与核心判断

3.1 常见问题

3.1.1 Spark DataSource 为什么会报 Hive Metastore 方法不存在

这次我一开始用 Spark SQL 跑：

DROP TABLE IF EXISTS demo_hudi_internal_orders;

实际报错：

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

在本文这套 Spark 3.5.6、Hive 4.0.1、Hudi 1.0.2 单机测试环境中，原因是 Spark 3.5.6 仍按它兼容的 Hive Metastore 客户端访问，而环境里的 Hive 是 4.0.1，Thrift API 行为不完全兼容。

本文修复方式是把验证代码改成 Spark DataSource，并显式设置：

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

如果要走 Hive Metastore，就要按 Spark、Hive、Hudi 的兼容矩阵重新校准依赖，而不是在业务 SQL 里硬绕。

3.1.2 `hoodie.cleaner.*` 参数为什么有警告

第一次写入时我沿用了旧参数，这是为了观察从旧配置迁移到 Hudi 1.0.2 新配置时的警告行为：

hoodie.cleaner.policy=KEEP_LATEST_COMMITS
hoodie.cleaner.commits.retained=2

Hudi 1.0.2 输出了警告：

The configuration key 'hoodie.cleaner.policy' has been deprecated and may be removed in the future.
Please use the new key 'hoodie.clean.policy' instead.
The configuration key 'hoodie.cleaner.commits.retained' has been deprecated and may be removed in the future.
Please use the new key 'hoodie.clean.commits.retained' instead.

所以后面的正式配置我改成：

hoodie.clean.policy=KEEP_LATEST_COMMITS
hoodie.clean.commits.retained=2

这类问题在 Hudi 版本升级时很常见。参数不是不能用，而是已经迁移，继续使用旧参数会增加后续升级成本。

3.1.3 MOR 表为什么只有 Log File 或查询变慢

如果 MOR 表更新很多，目录里会看到越来越多 .log 文件。原因是 update 走追加路径，没有立即合并到 Parquet。

处理思路：

确认是否是 MOR 表。
查看同一个 FileGroup 下 Log File 数量。
查看 Compaction 是否启用。
查看 Compaction 是否有 completed instant。
查看 Snapshot Query 是否有明显变慢。

不是所有 MOR 表都必须非常频繁地 Compaction。写入延迟和查询延迟要一起看。

3.1.4 Cleaner 是否可以直接设置成只保留 1 个 commit

不建议盲目这么做。

如果有长查询、增量任务、Flink 流读任务，太小的保留窗口可能导致下游还没读完，旧 FileSlice 已经被清理，结果出现文件找不到或增量断档。

更合理的做法是按最长查询时间和写入频率估算：

clean.commits.retained >= 最长查询耗时 / 平均 commit 间隔

例如每 10 分钟一个 commit，最长查询可能跑 1 小时，至少要保留 6 个 commit，再留一些余量。

3.2 读写机制的核心判断

Hudi 底层读写可以用三句话串起来：

Timeline 管事务：哪些写入完成、哪些写入还在进行、哪些表服务已经执行，都在 Timeline 里。
FileGroup 管版本：record key、partition path 和索引策略共同决定记录定位到哪个候选 FileGroup，Base File 和 Log File 一起构成某个时间点的 FileSlice。
Writer / Reader 管协调：Writer 负责写入和提交，Reader 根据 Timeline 构造一致视图，表服务负责把文件布局维护在可控范围内。

我以前刚接触 Hudi 时，很容易把它想成“带 update 的 Parquet”。后来更愿意把它理解成一套运行在文件系统上的表管理层：Parquet 只是 Base File，Log File 只是增量载体，真正让它成立的是 Timeline、Index、FileGroup、Payload 和 Table Services 这些机制组合在一起。

这也是 Hudi 的复杂度来源。它比普通 Hive 外表复杂很多，但换来的能力是更新、删除、事务、增量读取、流批协同和文件治理。

4. 总结

先把本文实验观察整理成一张表：

观察点	本文结果	说明
MOR 多次 upsert	生成 1 个 Base File 和多个 Log File	更新先进入同 FileGroup 的增量文件
Snapshot Query	能读到 `15001=FINISHED`	读取时合并 Base File 和 Log File
Timeline	看到 requested / inflight / completed deltacommit	读可见性主要看 completed instant
Compaction	本次未成功生成 plan	本文只记录调用结构和失败边界
Cleaner	成功运行但删除 0 个文件	小表没有足够多可安全回收的旧 FileSlice
Flink 写入原理	依赖 checkpoint 提交 completed instant	本节是前文实测链路的原理回顾

这篇主要把 Hudi 的底层读写逻辑串了一遍：

写入从 HoodieKey 开始，record key 和 partition path 决定记录定位。
preCombine 解决同 key 多版本输入的合并问题。
COW 更新重写 Base File，MOR 更新在常见路径下先进入 Log File。
Timeline 决定哪些 instant 对查询端可见。
Snapshot Query 读取 MOR 表时会合并 Base File 和 Log File。
Compaction 把 MOR 的读时合并提前成表服务合并。
Clean 负责回收旧 FileSlice，但保留窗口必须和查询、增量消费、流读延迟匹配。
Flink 写 Hudi 的关键是 checkpoint 提交 completed instant。

hnbian

https://www.hnbian.cn/posts/c3e5a7d9.html

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !

flink spark hudi compaction internals clean

Apache Hudi 学习笔记 15：Hive 集成 Hudi 与 Hive Sync 实战

2025-10-30 hudi

flink spark hive hudi mor query hive-sync metastore

Apache Hudi 学习笔记 13：Flink Hudi 高级能力详解

2025-10-08 hudi

flink hudi bucket changelog clustering catalog

Apache Hudi 学习笔记 14：Hudi 底层读写原理解析

1. 验证表和写入机制

1.1 背景问题

1.2 本文使用的验证表

1.3 写入内部流程

1.3.1 HoodieKey：记录定位的起点

1.4 去重与合并：preCombine 不只是去重

1.5 MOR 表写入：Base File 与 Log File 的关系

1.6 Timeline：写入状态不是文件名游戏

1.7 Flink 写入原理：checkpoint 提交 instant

2. 读取和表服务

2.1 读取内部流程

2.2 Compaction：把读时合并变成写后合并

2.2.1 本文 Compaction 未成功生成 plan 的边界情况

2.2.2 Compaction 配置与流读影响

2.3 Clean：删除旧 FileSlice，不是删除旧数据

2.4 流式读取：不是一直扫目录

3. 常见问题与核心判断

3.1 常见问题

3.1.1 Spark DataSource 为什么会报 Hive Metastore 方法不存在

3.1.2 hoodie.cleaner.* 参数为什么有警告

3.1.3 MOR 表为什么只有 Log File 或查询变慢

3.1.4 Cleaner 是否可以直接设置成只保留 1 个 commit

3.2 读写机制的核心判断

4. 总结

3.1.2 `hoodie.cleaner.*` 参数为什么有警告