Apache Hudi 实时入湖实践：写入方式与表服务

团队选定 Hudi 建设实时数据湖后，落地中高频出现三类困惑：同一张表用 Spark 写还是 Flink 写？MOR 表为什么越查越慢？Compaction 和 Clustering 到底谁管什么事、怎么配？本文按单引擎实施路径展开：先讲清表模型与索引，再对比写入方式，给出实时入湖链路搭建步骤，最后聚焦 Compaction、Clustering 与调优。

Hudi 表模型与关键概念

Hudi 提供两种表类型：

• Copy on Write（COW）：更新时直接重写整个数据文件，读时无需合并，读性能好；但每次更新都重写文件，写放大高。适合更新少、读密集的表。
• Merge on Read（MOR）：更新先写增量日志（log 文件），与基础 parquet 文件共存，读时合并；写性能好、适合高频更新，但 log 越多读越慢，需要 Compaction 定期把 log 合并进 parquet。

表数据按 record key（主键）与分区路径组织。写入时靠索引定位记录在哪个文件组，避免全表扫描。索引类型有 Simple Index（Spark 默认）、Bloom Index、HBase Index、Bucket Index（Flink 常用）。Timeline 记录每次操作（commit、delta commit、compaction、clustering、clean），支撑增量查询与 time travel。从 0.13.0 起 Hudi 还内置了 CDC 能力，可在表内记录变更前后镜像与操作类型，满足 GDPR 类记录级删除合规。

索引定位可以理解为两步：先按分区路径缩小范围，再在分区内用索引找到 record key 对应的文件组；找不到就视为新记录走插入路径。Simple Index 直接比较 key 与文件组内的记录，Bloom 用布隆过滤器先过滤再精查，Bucket 按 hash 字段取模固定映射到文件组。Flink 写入默认走状态索引，同样按主键维护记录到文件组的映射，这解释了为什么主键必须是稳定唯一的业务键——主键一旦可变，索引定位就会失效并产生重复数据。

写入方式选择：Spark 与 Flink 差异

Spark 写入以批量为主：upsert（按主键更新插入）、bulk_insert（大批量导入）、insert_overwrite。适合历史数据回填、离线批量入湖与重算。通过 DataFrame 或 Spark SQL 配置 hoodie.datasource.write.recordkey.field、precombine.field 等参数即可。

Flink 写入以流式为主：Append 与 Upsert 模式持续写入，配合 checkpoint 实现 exactly-once，是实时链路的主力。Flink CDC 到 Kafka 再到 Hudi 是当前最主流的实时入湖路径，业务库变更以分钟级甚至秒级延迟落到湖里。

选型权衡：实时性要求高、持续变更的场景选 Flink；大批量、可批量重算的历史场景选 Spark。同一个表尽量统一一种 writer 与索引策略——Spark 与 Flink 默认索引不同，混写同一张表容易出现重复记录或索引定位不一致，需要显式统一。

Spark 批量 upsert 的写法示例：

df.write.format("hudi")
  .option("hoodie.datasource.write.recordkey.field", "order_id")
  .option("hoodie.datasource.write.precombine.field", "ts")
  .option("hoodie.datasource.write.operation", "upsert")
  .mode(Append)
  .save("hdfs://namenode/warehouse/ods/orders")

历史回填建议用 bulk_insert 而非 upsert：bulk_insert 跳过索引定位直接批量写文件，导入更快，适合无更新语义的存量数据；有更新语义的数据用 upsert。

实时入湖链路搭建

典型链路：业务库 → Flink CDC 捕获变更 → Kafka 缓冲 → Flink 写 Hudi → Hudi 表（HDFS 或对象存储）→ 下游 Spark、Trino、Presto 查询。Kafka 的作用是削峰与解耦，避免 CDC 直接打到湖上造成写放大。

建表用 Flink SQL，示例：

CREATE TABLE orders_hudi (
  order_id BIGINT,
  user_id BIGINT,
  amount DECIMAL(10,2),
  ts TIMESTAMP(3),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'hudi',
  'path' = 'hdfs://namenode/warehouse/ods/orders',
  'table.type' = 'MERGE_ON_READ',
  'hoodie.datasource.write.recordkey.field' = 'order_id',
  'hoodie.datasource.write.precombine.field' = 'ts',
  'compaction.async.enabled' = 'true',
  'compaction.trigger.strategy' = 'num_commits',
  'compaction.delta_commits' = '5'
);

关键设计点：主键要唯一稳定（业务主键而非自增代理键）；precombine 字段用于乱序与后写覆盖；分区按查询模式设计（通常按时间）；checkpoint 间隔决定数据可见延迟与故障恢复粒度。

链路搭建分五步：第一步在源库开启 CDC 所需配置（如 Binlog、归档与补充日志）；第二步建 Kafka topic 并确认分区数足够并行消费；第三步写 Flink CDC 作业把变更同步到 Kafka；第四步写 Flink Hudi 作业消费 Kafka 写入湖表；第五步注册 Hive 或湖仓元数据供查询引擎访问。每一步都要独立验证：CDC 能捕获增删改、Kafka 不丢消息、Hudi 表能查到最新数据。先串通一条小表再放大，是避免链路问题与表问题混在一起的最快方式。

Compaction 与 Clustering 配置

Compaction 只针对 MOR 表，把 log 文件合并进基础 parquet，控制读放大。触发策略常用 num_commits（默认 5 个 commit 合并一次）或 time_elapsed（按时间）。建议开启异步 Compaction，避免阻塞写入：compaction.async.enabled=true，compaction.tasks 控制并行度。对 CDC 高频小事务场景，log 增长快，可以把 delta_commits 调低（如 3）；对低频更新表可以调高，减少资源消耗。

理解 MOR 的读视图有助于定位"越查越慢"：MOR 表对外提供两种读模式，Read Optimized 只扫基础 parquet 文件、速度快但不包含最新 log 数据；Snapshot 实时合并 parquet 与 log、数据最新但 log 越多越慢。业务上延迟要求不高的查询可以走 Read Optimized，实时报表才走 Snapshot，两种视图配合能显著摊薄读放大，而不是把所有压力都堆在 Snapshot 上。

Clustering 是文件布局优化，把大量小文件合并成接近目标大小的大文件，并可对列排序。Spark 侧 hoodie.clustering.inline.max.commits=4 或 hoodie.clustering.async.enabled=true；Flink 侧同样支持异步 clustering。Clustering 重写文件会产生新版本，需要 Cleaning 定期回收旧版本，两者要一起看，否则存储只增不减。

小文件与性能调优

小文件是 Hudi 最常见的性能杀手：高频小批量写入、分区过细、并发写入都会产生大量小文件，拖垮元数据服务与查询扫描。

控制手段按优先级：

1. 写入时控制：hoodie.parquet.small.file.limit（默认 100MB）让新数据优先追加进已有小文件，hoodie.parquet.max.file.size（默认 120MB）限制单文件大小。
2. 分区粒度：按查询模式定，避免分钟级分区。
3. 定时 Clustering：兜底合并存量小文件。
4. 合理批量：Flink 侧攒批写入（批量大小、checkpoint 间隔）减少 commit 频率。

查询侧调优：MOR 表 log 多时读放大明显，及时 Compaction；启用列剪枝与增量查询（incremental read）减少扫描；存储侧用列式压缩（parquet 加 snappy 或 zstd）。对象存储部署要注意 Hudi 对原子提交与 rename 的要求，优先选择支持原子语义的存储，避免并发写冲突。

运维上建议给表服务加监控：观察每张表每小时的 commit 数、log 文件数、小文件占比、compaction 与 clustering 的积压量。commit 数激增往往意味着批量过小；log 数持续上涨说明 compaction 跟不上；小文件占比高说明需要调 small.file.limit 或触发 clustering。把这些指标与查询延迟关联起来，就能在用户反馈前发现表服务失衡。

常见坑

• Spark 与 Flink 混写同一张表导致主键重复：统一 writer 与索引策略，或明确两段式（历史用 Spark、实时用 Flink）并做好衔接校验。
• Compaction 跟不上写入：log 无限增长，查询越来越慢，表现为读放大；要监控 log 文件数量与大小，动态调整 delta_commits 与 compaction.tasks。
• 小文件爆炸：只开写入不配 Clustering，文件数只增不减；把 small.file.limit 与异步 Clustering 一起配齐。
• Flink exactly-once 依赖 checkpoint：作业无状态恢复或 checkpoint 失效时会重复写，靠 precombine 字段与主键去重兜底。
• 分区策略变更代价高：Hudi 改分区字段需要重写数据，前期按数据量增长规划好分区粒度。

下一步

先选一张高频变更的明细表做试点：用 Flink CDC 写入 MOR 表，开启异步 Compaction（delta_commits=5）与 Clustering，监控 log 文件数与查询延迟；跑通实时链路后，用 Spark bulk_insert 补齐历史回填，最后把表服务参数按高频更新、低频更新、批量加载三类沉淀为模板。