4. Apache Hudi 深度解析（二）：写入流程、增量查询、Clustering 与 Compaction

上一章我们聊了 Hudi 的表类型和索引，算是打了个底。这一章咱们直接上手，看看数据到底是怎么写进去的，又怎么读出来。特别是增量查询、Clustering 和 Compaction 这几个核心机制，我当年在生产环境里可没少跟它们打交道。

4.1 写入流程：从 Record 到 File Group

Hudi 的写入，说白了就是把你的数据记录，组织成一个个文件组。我习惯把整个过程拆成三步：

  1. 输入转换：把上游的 Avro、Parquet 或 JSON 数据，转成 Hudi 内部的 HoodieRecord。
  2. 索引定位：根据记录的主键，去索引里查一下——这条数据之前有没有写过？
  3. 文件写入：根据索引结果，决定是插入新文件，还是更新已有文件。

举个例子，你有一条用户数据，主键是 user_id=1001。Hudi 会先查索引，如果发现这个 ID 已经存在，就把这条记录标记为「更新」，写入对应的文件组。如果是新 ID，就标记为「插入」，分配到一个新的文件组。

关键点：写入时，数据先写到内存中的缓冲区（类似 Spark 的 Dataframe），等攒够一批（比如 120MB），再一次性刷到磁盘。这样做能大幅减少小文件数量。

嗯，这里要注意写入模式的选择。Hudi 支持三种：

写入模式适用场景我的经验
UPSERT既有插入又有更新最常用，但索引开销大
INSERT纯新增数据适合日志导入，速度快
BULK_INSERT大规模初始加载我用来做历史数据回填，比 INSERT 快 3-5 倍

小技巧：如果你确定数据里没有重复主键，直接用 BULK_INSERT。它跳过了索引查找和排序，写入性能会好很多。我在一次 10TB 历史数据迁移中，就靠这个把时间从 6 小时压缩到了 2 小时。

4.2 增量查询：只读变化的数据

增量查询是 Hudi 最吸引我的特性之一。你想想看，传统数仓里你要查「今天新增了哪些数据」，要么全表扫描，要么依赖业务时间戳。Hudi 的做法更优雅——它记录了每个 commit 变更的文件列表。

查询时，你只需要指定一个起始时间戳：

-- 查询从 '20240101000000' 之后的所有变更
SELECT * FROM hudi_table
WHERE _hoodie_commit_time > '20240101000000'

为什么能做到？因为 Hudi 的元数据里维护了一张 Timeline，记录了每次写入的 commit 信息。增量查询本质上就是去 Timeline 里找「从上次查询到现在，哪些文件发生了变化」，然后只读这些文件。

避坑指南：我曾经遇到过一个坑——增量查询的起始时间戳，必须是 Hudi 的 commit 时间，不是你业务数据里的时间。如果你用业务时间戳去过滤，结果会不准确。因为 Hudi 的 commit 时间和业务时间可能差好几秒。

增量查询的典型场景：

  • 实时 ETL 管道：每 5 分钟拉取一次增量数据
  • CDC 同步：从 Hudi 表同步变更到下游系统
  • 数据回填修正：只修复某段时间内写错的数据

4.3 Clustering：把小文件合并成大文件

写入多了，小文件问题就来了。你想想看，HDFS 上每个小文件都要占用一个 block 的元数据，NameNode 压力大，查询性能也差。Clustering 就是专门解决这个问题的。

它的核心逻辑很简单：把多个小文件组（File Group）合并成一个大文件组。但合并的时候有个讲究——不能破坏数据的一致性。

Hudi 的做法是：

  1. 先选出需要合并的小文件（比如小于 128MB 的）
  2. 把这些文件的数据读出来，重新排序、合并
  3. 写入新的大文件，同时更新元数据
  4. 旧的小文件标记为「已废弃」，等待清理

注意：Clustering 是异步执行的，不会阻塞正常的写入和查询。我习惯在业务低峰期（比如凌晨 2 点）触发 Clustering，避免影响在线服务。

配置 Clustering 的几个关键参数：

参数默认值说明
hoodie.clustering.inlinefalse是否在写入时同步执行
hoodie.clustering.plan.strategy.target.file.max.bytes1073741824 (1GB)合并后的目标文件大小
hoodie.clustering.plan.strategy.small.file.limit314572800 (300MB)小于此值的文件才参与合并

我的建议：目标文件大小设成 HDFS block size 的整数倍（比如 256MB 或 512MB），这样查询时能充分利用数据本地性。我踩过坑，设成 1GB 后，Map 任务数变少，但单个任务处理时间变长，反而得不偿失。

4.4 Compaction：MOR 表的救星

如果你用的是 MOR（读时合并）表，那 Compaction 就是你绕不开的话题。MOR 表把数据分成两部分：Base 文件（Parquet，列存，压缩率高）和 Log 文件（Avro，行存，写入快）。

Compaction 的作用，就是把 Log 文件里的增量数据，合并到 Base 文件里。说白了，就是把行存转成列存，提升查询性能。

我画了一张图，帮你理解 Compaction 的流程：

Compaction 流程示意图 写入请求 写入 Log 文件 Log 文件累积 触发条件 Log 大小 / 时间间隔 读取 Log + Base 合并生成新 Base 更新元数据 清理旧 Log 文件 说明：Compaction 是异步的，不会阻塞写入。但查询时如果 Log 文件太多， 读性能会下降。建议根据写入频率，合理设置 Compaction 触发间隔。

Compaction 有两种触发方式：

  • 内联 Compaction：写入时同步执行，简单但影响写入性能
  • 异步 Compaction：单独起一个 Spark 任务执行，我推荐这种方式

我曾经踩过的坑：有一次我把 Compaction 的触发间隔设得太长（24 小时），结果 Log 文件累积了上百个。查询时，Hudi 需要合并所有 Log 文件，一个简单的 count(*) 查询跑了 20 分钟。后来我把间隔改成 1 小时，或者 Log 文件总大小超过 500MB 就触发，问题就解决了。

Compaction 策略的选择也很重要：

策略特点适用场景
NUM_COMMITS按 commit 次数触发写入频率稳定时
TIME_ELAPSED按时间间隔触发写入频率波动大时
NUM_IO_ERROR按 IO 错误次数触发不太常用，我基本不用

我的配置建议：对于大多数场景，我习惯用 TIME_ELAPSED + NUM_COMMITS 组合。比如每 30 分钟或每 5 个 commit 触发一次 Compaction。这样既能保证 Log 文件不会太多，又不会频繁触发浪费资源。

4.5 小结一下

这一章我们聊了 Hudi 的写入流程、增量查询、Clustering 和 Compaction。说白了，写入流程决定了数据怎么进去，增量查询决定了怎么高效读变化，Clustering 和 Compaction 则是保证系统长期稳定运行的关键。

我个人觉得，理解这几个机制的核心，不在于记住每个参数，而在于明白它们为什么存在。比如 Compaction，它存在的根本原因是 MOR 表的设计取舍——用写入性能换查询性能。你理解了这一点，配置参数时就不会盲目。

← 上一章 📖 返回目录 下一章 →