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

前面几篇已经把 Flink 写 Hudi 的基础链路、参数调优、CDC 入湖都跑过了。到这里，Flink Hudi 已经不只是“建表、写入、查询”这么简单了。继续往生产场景走，会遇到更多高级能力：Changelog、Append 模式、Clustering、Bucket Index、Hudi Catalog。

这些能力看起来分散，其实都围绕一个问题：实时写入数据湖以后，怎样降低状态成本、控制文件数量、改善查询性能、管理元数据，并尽量保留 CDC 的变化语义。

1. 高级能力模型

1.1 背景问题

Flink 持续写 Hudi 表以后，问题会逐渐从“写不写得进去”变成：

1. CDC 变化能不能被下游按 changelog 方式消费。
2. 纯追加数据能不能跳过不必要的索引开销。
3. 小文件越来越多以后，如何重新组织文件。
4. 大状态索引会不会拖慢 checkpoint。
5. 多张 Hudi 表能不能用 Catalog 统一管理。
6. MOR 表为什么只有 log 文件。
7. 写入后为什么存储目录一时看不到数据。

我对这篇的定位是：把 Flink Hudi 常见高级能力放到一个整体里理解，而不是把每个参数孤立背下来。

先用一张图把这些能力放到同一条 Flink Hudi 写入链路里：

这张图主要说明 Changelog、Append、Clustering、Bucket Index 和 Hudi Catalog 分别作用在 Flink Hudi 写入链路的变化语义、写入模式、文件治理、索引定位和元数据管理环节。

可以看到，这些能力并不是互相孤立的参数。Changelog 解决变化语义，Append 降低纯追加写入成本，Clustering 负责文件治理，Bucket Index 降低大状态压力，Hudi Catalog 负责元数据和多表管理。它们最终都围绕 Flink 写 Hudi 表时的 checkpoint、state、writer 和 table service 协同工作。

1.2 Changelog 模式

CDC 链路里，数据不是普通 append，而是变化流。Flink 内部会把变化表示成几类 row kind：

Row Kind	含义
+I	insert
-U	update before
+U	update after
-D	delete

Hudi 的 Changelog 模式可以让下游在某些场景下读取变化流，而不只是读取最新快照；但它不等同于数据库 binlog，也不保证在所有 compaction、clean 或 reader 延迟场景下完整保留每一次中间变化。

这张图主要说明 Flink changelog row kind 进入 Hudi 后，能否被下游再次按变化流读取，取决于表类型、changelog 配置、compaction 和 clean 保留策略。

下面是配置形态示例，本文没有对 changelog 读取结果做完整实测；真实使用前需要用当前 Hudi 1.0.2 Flink connector 单独验证 row kind 输出、compaction 后保留情况和 clean 保留窗口。本文以 Hudi 1.0.2 Flink connector 的 changelog.enabled 为准，其他版本需要按对应配置文档确认。

CREATE TABLE hudi_orders_changelog (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_id STRING,
  amount DECIMAL(10,2),
  status STRING,
  ts BIGINT,
  dt STRING
) PARTITIONED BY (dt)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs:///hudi/orders_changelog',
  'table.type' = 'MERGE_ON_READ',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts',
  'changelog.enabled' = 'true'
);

读取时可以配合 Flink SQL Client 的 tableau 模式观察 row kind：

SET 'sql-client.execution.result-mode' = 'TABLEAU';
SET 'table.dynamic-table-options.enabled' = 'true';

SELECT order_id, status, dt
FROM hudi_orders_changelog
/*+ OPTIONS(
  'read.streaming.enabled'='true',
  'read.streaming.check-interval'='4',
  'read.start-commit'='earliest'
) */;

Changelog 要注意几个限制：

1. 它更适合 CDC 链路，不适合普通批量快照表。
2. 对 MOR 表更常见，因为未 compaction 前 Log File 中更可能保留变化信息；但 compaction 后中间变更可能被合并，仍然不能当成完整审计日志。
3. Compaction 和 clean 会影响历史变化保留。
4. 很多场景下它是 Best Effort，不要把它当成数据库 binlog 的完整替代。

Best Effort 这点尤其重要。异步 compaction 可能会把多条中间变更合并成最后一条，流读消费不够及时，下游就只能读到压缩后的最终状态。除了 compaction，clean 和 timeline archive 也会影响历史变化能否被落后的 reader 继续读取。可以通过调大 compaction 触发间隔给 reader 留 buffer：

WITH (
  'compaction.delta_commits' = '5',
  'compaction.delta_seconds' = '3600'
)

这里是参数方向示例，具体 key、单位和默认值以 Hudi 1.0.2 Flink Configurations 为准。这不是“保证永远保留 changelog”，只是延长中间变更被合并前的窗口。需要审计级回放时，Kafka CDC Topic 仍然要保留。

这里可以理解为：如果下游真的需要完整审计级别的变更历史，Kafka CDC 原始 Topic 仍然应该保留。Hudi Changelog 更适合在湖表消费链路中传递变化，而不是完全替代 CDC 日志。

1.3 Append 模式

本文说的 Append 模式，主要指 Hudi Flink 中使用 write.operation='insert' 写纯追加数据，不是 Flink 动态表语义里所有 append-only source 的总称。它适合只追加、不更新的数据，比如埋点日志、行为日志、传感器事件。如果数据天然不会更新，用 upsert 反而会引入额外索引成本。

在 Flink Hudi 中，Append 写入通常可以用：

'write.operation' = 'insert'

Append 下 COW 和 MOR 的文件行为不一样：

表类型	INSERT / Append 行为	需要关注的问题
COW	直接写新的 Parquet Base File	小批量写入容易产生小文件，通常配合 Clustering
MOR	可能写入 Log File，也可能在 small file 策略和当前文件状态下写出 / 填充 Base File	查询要关注 Snapshot 合并和 Compaction，不能简单假设一定只有 Log File

所以 Append 不等于“没有表服务”。它只是跳过了 update 定位成本，文件治理仍然要靠 clustering、compaction 和合理的写入批次。

本文实际创建了一张 COW Append 表：

这里的 dt 是模拟业务分区日期，用于演示多分区写入和文件布局，不代表文章实际写作日期；ts 是模拟业务事件时间，用作 precombine.field 排序字段。write.insert.cluster=true 是 Hudi 1.0.2 Flink connector 中用于 insert 写入阶段小文件合并 / 文件组织的参数，具体行为以当前版本文档和实测为准。

CREATE TABLE demo_hudi_append_orders (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_id STRING,
  amount DECIMAL(10,2),
  status STRING,
  ts BIGINT,
  dt STRING
) PARTITIONED BY (dt)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs:///hudi/demo_hudi_append_orders',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'insert',
  'precombine.field' = 'ts',
  'hoodie.metadata.enable' = 'false',
  'write.tasks' = '1',
  'write.bucket_assign.tasks' = '1',
  'read.tasks' = '1',
  'write.insert.cluster' = 'true',
  'write.rate.limit' = '1000'
);

即便是 insert / append 场景，Hudi 表仍然需要 record key。第一次我没有定义 record key，关键报错摘录如下：

org.apache.hudi.exception.HoodieValidationException:
Primary key definition is required, the default primary key field 'uuid'
does not exist in the table schema, use either PRIMARY KEY syntax or option
'hoodie.datasource.write.recordkey.field' to specify.

所以在本文 Hudi 1.0.2 Flink SQL 环境中，即使 write.operation='insert'，Hudi 表仍然需要能够构造 record key；Flink SQL 里通常通过 PRIMARY KEY NOT ENFORCED 或对应 record key 配置声明。这里的 PRIMARY KEY NOT ENFORCED 不是数据库强约束。

所以我后来在 DDL 中显式写了：

order_id BIGINT PRIMARY KEY NOT ENFORCED

写入数据：

INSERT INTO demo_hudi_append_orders VALUES
  (13001, 'u301', CAST(59.90 AS DECIMAL(10,2)), 'CREATED', 1759816200000, '2025-10-07'),
  (13002, 'u302', CAST(79.00 AS DECIMAL(10,2)), 'CREATED', 1759816260000, '2025-10-07'),
  (13003, 'u303', CAST(129.00 AS DECIMAL(10,2)), 'CREATED', 1759816320000, '2025-10-08');

输出：

[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 0a3b9743894a91fda7e1208ea7c586c9

Timeline：

Found 4 items
-rw-r--r--   1 root supergroup          0 2025-10-08 13:51 /hudi/demo_hudi_append_orders/.hoodie/timeline/20251008135158440.commit.requested
-rw-r--r--   1 root supergroup          0 2025-10-08 13:51 /hudi/demo_hudi_append_orders/.hoodie/timeline/20251008135158440.inflight
-rw-r--r--   1 root supergroup       4376 2025-10-08 13:52 /hudi/demo_hudi_append_orders/.hoodie/timeline/20251008135158440_20251008135200431.commit
drwxr-xr-x   - root supergroup          0 2025-10-08 13:51 /hudi/demo_hudi_append_orders/.hoodie/timeline/history

在 Hudi 1.0.x 中，完成态 Instant 文件名里可以看到 begin instant 和 completion instant，例如 begin_completion.commit；这里前半段更接近请求 / 开始时间，后半段是完成时间。requested / inflight 用于表达 Instant 状态流转；判断已经成功提交的动作时，主要看完成态 commit 文件。

查询分区计数：

+------------+-----+
|         dt | cnt |
+------------+-----+
| 2025-10-07 |   2 |
| 2025-10-08 |   1 |
+------------+-----+
2 rows in set (5.20 seconds)

1.4 Clustering

实时写入 Hudi 表以后，小文件和文件布局是绕不开的问题。尤其是 Kafka CDC、小批量 upsert、按小时或按城市细分分区时，文件容易越来越碎。

Clustering 的作用是重写文件布局：

1. 合并小文件。
2. 按指定列重新排序。
3. 生成更适合查询的 FileGroup。
4. 降低后续查询的文件打开成本。

这张图主要说明 clustering 不是改变业务数据，而是把已有 FileGroup / FileSlice 重新组织成更适合查询的文件布局。

下面是 Hudi 1.0.2 Flink clustering 配置形态示例，本文没有完整实测 clustering plan 生成和执行；具体 key、单位和默认值需要以当前版本文档与实测为准。

Insert 写入阶段文件聚合示例：

CREATE TABLE demo_hudi_inline_cluster (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_id STRING,
  amount DECIMAL(10,2),
  status STRING,
  ts BIGINT,
  dt STRING
) PARTITIONED BY (dt)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs:///hudi/demo_hudi_inline_cluster',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'insert',
  'precombine.field' = 'ts',
  'write.insert.cluster' = 'true'
);

Async Clustering 示例：

CREATE TABLE demo_hudi_async_cluster (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_id STRING,
  amount DECIMAL(10,2),
  status STRING,
  ts BIGINT,
  dt STRING
) PARTITIONED BY (dt)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs:///hudi/demo_hudi_async_cluster',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts',
  'clustering.schedule.enabled' = 'true',
  'clustering.async.enabled' = 'true',
  'clustering.tasks' = '2',
  'clustering.plan.strategy.target.file.max.bytes' = '134217728',
  'clustering.plan.strategy.small.file.limit' = '104857600',
  'clustering.plan.strategy.sort.columns' = 'dt,order_id'
);

Inline 和 Async 的差异：

方式	特点	适合场景
Inline Clustering	写入路径同步整理	小规模写入、初始化、对延迟不敏感
Async Clustering	后台异步整理	长期实时写入、需要降低写入延迟

write.insert.cluster 更偏 insert 写入阶段的小文件合并 / 文件组织，不应简单等同于 Hudi clustering table service。生产实时链路更常见的思路是先让写入任务稳定，再通过异步 clustering 或独立表服务逐步整理文件布局。

还有一种做法是离线 Clustering：写入任务只负责定期 schedule clustering plan，真正执行交给独立 Flink 作业。这样资源隔离更好，适合写入吞吐已经很高、在线 clustering 会影响 checkpoint 的场景。

写入表可以这样配置：

WITH (
  'clustering.schedule.enabled' = 'true',
  'clustering.async.enabled' = 'false'
)

然后由独立作业执行 clustering plan。Flink 侧能否独立执行 clustering、入口类名称是什么，需要以 Hudi 1.0.2 文档和当前 hudi-flink1.20-bundle-1.0.2.jar 实际内容为准。本文当前机器没有该 bundle 可供 jar tf 核对，所以这里只保留排查方向，不把它当成已验证命令：

jar tf /opt/module/flink-1.20.1/lib/hudi-flink1.20-bundle-1.0.2.jar \
  | grep -i ClusteringJob

如果当前 Hudi 1.0.2 Flink bundle 不支持独立执行 clustering，只能 schedule plan，则应使用 Spark clustering 作业或当前版本支持的表服务作业执行。非分区表和分区表的 clustering plan 生成逻辑可能不同，生产前都需要单独验证。

1.5 Bucket Index

普通索引需要维护 key 到 FileGroup 的映射。Flink 写 Hudi 时，这部分映射可能进入 state，表越大，state 成本越高。对于 SIMPLE Bucket Index，可以先粗略理解为用固定 hash 桶减少定位成本：

bucket = hash(recordKey) % numBuckets

如果使用 CONSISTENT_HASHING bucket engine，桶数和路由逻辑会更复杂，需要结合 clustering / 扩缩桶能力单独理解。

这张图主要说明 Bucket Index 通过 hash(record key) 把记录路由到固定 bucket / FileGroup，从而减少 Flink state 中保存全量 key 映射的压力。

本文实际创建了一张 Bucket Index 表：

CREATE TABLE demo_hudi_bucket_orders (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_id STRING,
  amount DECIMAL(10,2),
  status STRING,
  ts BIGINT,
  dt STRING
) PARTITIONED BY (dt)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs:///hudi/demo_hudi_bucket_orders',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts',
  'hoodie.metadata.enable' = 'false',
  'index.type' = 'BUCKET',
  'hoodie.bucket.index.engine' = 'SIMPLE',
  'hoodie.bucket.index.hash.field' = 'order_id',
  'hoodie.bucket.index.num.buckets' = '4',
  'write.tasks' = '1',
  'write.bucket_assign.tasks' = '1',
  'read.tasks' = '1'
);

本文使用 SIMPLE Bucket Index，不展开 consistent hashing bucket。hash.field 可以是 record key 的子集，本文用 order_id 作为 record key 和 bucket hash field，实测写入可以通过。这里的 dt 仍然是模拟业务分区日期，ts 是模拟业务事件时间和 precombine 排序字段，用于和 Append 示例保持一致。

写入数据：

INSERT INTO demo_hudi_bucket_orders VALUES
  (14001, 'u401', CAST(19.90 AS DECIMAL(10,2)), 'CREATED', 1759816400000, '2025-10-07'),
  (14002, 'u402', CAST(88.00 AS DECIMAL(10,2)), 'CREATED', 1759816460000, '2025-10-07'),
  (14002, 'u402', CAST(88.00 AS DECIMAL(10,2)), 'PAID', 1759816600000, '2025-10-07'),
  (14003, 'u403', CAST(268.00 AS DECIMAL(10,2)), 'CREATED', 1759816660000, '2025-10-08');

输出：

[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 5f0b90c8325620d3b98550c9920d7817

Timeline：

Found 4 items
-rw-r--r--   1 root supergroup          0 2025-10-08 13:52 /hudi/demo_hudi_bucket_orders/.hoodie/timeline/20251008135202013.commit.requested
-rw-r--r--   1 root supergroup          0 2025-10-08 13:52 /hudi/demo_hudi_bucket_orders/.hoodie/timeline/20251008135202013.inflight
-rw-r--r--   1 root supergroup       4567 2025-10-08 13:52 /hudi/demo_hudi_bucket_orders/.hoodie/timeline/20251008135202013_20251008135203060.commit
drwxr-xr-x   - root supergroup          0 2025-10-08 13:51 /hudi/demo_hudi_bucket_orders/.hoodie/timeline/history

这里的 Timeline 同样是 Hudi 1.0.x 布局：完成态 commit 文件名包含 begin instant 和 completion instant，requested / inflight 只用于表达状态流转。

查看文件时可以看到 fileId 前缀中带有类似 bucket id 的编号：

-rw-r--r--   1 root supergroup     435771 2025-10-08 13:52 /hudi/demo_hudi_bucket_orders/2025-10-07/00000001-1aa9-44b7-9a3d-a0bfab941cb4_0-1-2_20251008135202013.parquet
-rw-r--r--   1 root supergroup     435748 2025-10-08 13:52 /hudi/demo_hudi_bucket_orders/2025-10-07/00000002-3ca0-490f-ad67-f4d400c57d29_0-1-2_20251008135202013.parquet
-rw-r--r--   1 root supergroup     435770 2025-10-08 13:52 /hudi/demo_hudi_bucket_orders/2025-10-08/00000003-c50d-4023-a73c-f80b940ec645_0-1-2_20251008135202013.parquet

文件名前半段是 fileId / bucket 相关标识，中间的 0-1-2 类似写入任务相关 token，最后的 20251008135202013 是写入 instant，不要把整个文件名都理解成 bucket id。

查询结果：

+----------+---------+------------+
| order_id |  status |         dt |
+----------+---------+------------+
|    14001 | CREATED | 2025-10-07 |
|    14002 |    PAID | 2025-10-07 |
|    14003 | CREATED | 2025-10-08 |
+----------+---------+------------+
3 rows in set (0.90 seconds)

14002 最终是 PAID，说明 Bucket Index 下 upsert 语义正常。

Bucket Index 的优点：

1. 降低索引 state 成本。
2. 定位 FileGroup 更直接。
3. 适合大表和高吞吐写入。

限制也很明显：

1. 对于 SIMPLE Bucket Index，bucket 数设置后不适合随意变更；consistent hashing bucket 方向具备更复杂的动态调整能力，但需要结合对应版本、clustering 和写入限制单独验证。
2. hash 分布不均可能导致热点 bucket。
3. 分区字段变化、跨分区更新仍然要谨慎处理。

和默认 Flink State Index 对比，可以这样理解：

对比项	Flink State Index	Bucket Index
key 定位方式	state 保存 record key 到 fileId 的映射	hash(key) % bucket_num 定位 FileGroup
state 成本	表越大，state 压力越明显	通常不需要像 Flink State Index 那样保存完整 key -> fileId 映射，但仍需要维护 bucket / file group 相关文件视图和写入状态
扩展性	可以跟随文件大小和 state 扩展	SIMPLE bucket 数不适合随意改
跨 partition 更新	配合 index.global.enabled 可以处理	分区变化要谨慎，CDC 顺序和分区字段要稳定

如果同一个 record key 的 partition path 会变化，Bucket Index 是否能正确清理旧分区记录，需要按 Hudi 1.0.2 Flink connector、index.global.enabled、update partition path 相关能力实测确认。如果业务 key 分布极不均匀，Bucket Index 也会把热点 key 压到少数 bucket 上。建表前最好先按历史数据估算 key 分布，而不是只按总数据量拍一个 bucket 数。

1.6 Hudi Catalog

前面我们一直使用普通 Flink default catalog，然后在 DDL 里写 connector='hudi' 和 path。表多了以后，这种方式不够集中。Hudi Catalog 可以统一管理 Hudi 表。

这张图主要说明 Hudi Catalog 负责统一管理表定义、默认路径和元数据入口，不等同于 Hudi Timeline 本身。

DFS Catalog 示例：

CREATE CATALOG hudi_dfs WITH (
  'type' = 'hudi',
  'mode' = 'dfs',
  'catalog.path' = 'hdfs:///hudi/catalog'
);

本文使用 Hudi 1.0.2 Flink Catalog 的 dfs mode 参数写法，其他版本需要按文档调整。本文只验证 DFS Catalog 能被 SQL Client 创建和切换；Catalog 建表和已有表纳管会在后续单独验证。

本文实测命令：

SET 'execution.runtime-mode' = 'batch';
SET 'sql-client.execution.result-mode' = 'TABLEAU';

CREATE CATALOG hudi_dfs WITH (
  'type' = 'hudi',
  'mode' = 'dfs',
  'catalog.path' = 'hdfs:///hudi/catalog'
);

SHOW CATALOGS;
USE CATALOG hudi_dfs;
SHOW DATABASES;

输出：

Flink SQL> +-----------------+
|    catalog name |
+-----------------+
| default_catalog |
|        hudi_dfs |
+-----------------+
2 rows in set

Flink SQL> +---------------+
| database name |
+---------------+
|       default |
+---------------+
1 row in set

如果要使用 HMS Catalog，可以配置 Hive Metastore：

CREATE CATALOG hudi_hms WITH (
  'type' = 'hudi',
  'mode' = 'hms',
  'hive.conf.dir' = '/opt/module/hive/conf'
);

DFS mode 更轻量，适合本地测试和文件系统环境；HMS mode 更适合和 Hive / Spark / Trino 共享元数据。HMS mode 需要单独验证 Flink、Hive Metastore、Hudi Catalog 和 Hive 依赖版本兼容性。本文前面的单机环境使用 Hive 4.0.1，已经出现过部分 Spark/HMS 兼容问题，因此这里不要直接把 HMS 示例当成已验证生产配置。/opt/module/hive/conf 按前文软链接目录书写，如果你的环境使用 /opt/module/hive-4.0.1/conf，按实际路径替换。

Catalog 还可以放在 SQL Client 初始化文件里，这样每次进入 SQL Client 不用重复创建：

cat > /opt/module/flink-1.20.1/conf/sql-client-init-hudi.sql <<'SQL'
CREATE CATALOG hudi_dfs WITH (
  'type' = 'hudi',
  'mode' = 'dfs',
  'catalog.path' = 'hdfs:///hudi/catalog'
);

USE CATALOG hudi_dfs;
SQL

/opt/module/flink-1.20.1/bin/sql-client.sh \
  -i /opt/module/flink-1.20.1/conf/sql-client-init-hudi.sql

SQL Client 初始化文件参数以 Flink 1.20.1 当前命令帮助为准，可通过 sql-client.sh --help 确认。

某些版本的 Hudi Catalog 支持 table.external 选项，具体默认值和行为以当前版本文档为准。我一般这样区分：

选项	含义
table.external=false	Catalog 管理表属性和默认表路径，适合新建受管表
table.external=true	更偏向引用已有外部表路径，适合纳管已有 Hudi 表

如果使用 HMS mode，Catalog 可以继承 Hive 相关公共配置，减少每张表重复写 hive sync 参数。

2. 常见问题与使用边界

2.1 常见基础问题

2.1.1 存储目录一直看不到数据

Flink Hudi sink 依赖 checkpoint 提交 commit。只看到 Flink job RUNNING 不代表 Hudi commit 已经完成。先看 Timeline：

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

如果只有 requested / inflight，没有 completed commit，要继续看 checkpoint 是否成功、slot 是否足够、source 是否还在运行。

Flink writer 刷数据到磁盘常见有三种触发方式：

1. 单个 bucket 的内存积累到阈值。
2. writer 总 buffer 积累到阈值。
3. checkpoint 触发，把内存里的数据 flush 出去并推进 commit。

所以小数据量测试时，最常见的原因其实是 checkpoint 没成功，或者 source 已经结束但没有触发可提交的 checkpoint。

对于 INSERT VALUES 这类有界输入，作业结束也可能推进最终提交；对于 Kafka CDC 这类无界流，checkpoint 是否持续成功才是 Hudi commit 稳定推进的关键。

2.1.2 数据重复

常见原因：

1. 没有定义 PRIMARY KEY。
2. precombine.field 不正确。
3. 使用 insert 写入了本应 upsert 的 CDC 数据。
4. 分区字段变化但索引策略没有处理跨分区更新。
5. 全量接增量时没有做 index bootstrap。
6. index.state.ttl 太短，长周期回补时旧 key 映射已经过期。

对于 CDC 表，默认优先使用：

'write.operation' = 'upsert',
'precombine.field' = 'ts'

Hudi 1.0.2 的 Flink State Index 默认 index.global.enabled=true、index.state.ttl=0，表示默认支持旧分区更新且索引永久保存，这里以 Hudi 1.0.2 配置文档为准。为了性能，有些线上表会手动给 index.state.ttl 设一个天数，这时如果回补一个月前的数据，就要确认 TTL 没把历史 key 清掉。

另外，write.precombine 控制写入前 buffer 阶段是否先去重。Hudi 1.0.2 默认值是 false，原因是插入和 MOR upsert 场景为了性能会接受更多中间数据，后续在读或 merge 阶段处理。CDC 场景如果上游乱序明显，不能只依赖这个开关，关键还是选对 precombine.field。

2.1.3 MOR 表只有 log 文件

MOR 表写入更新时追加 Log File 是正常现象。只有经过 compaction 后才会生成新的 Base File。要看 compaction 是否触发，而不是看到 log 文件就判断异常。

可以先查日志或 Timeline 中的 compaction 记录。如果看到类似 After filtering, Nothing to compact 的日志，说明当前 strategy 暂时没有选择出需要压缩的 FileSlice；具体日志文本会随 Hudi 版本变化。MOR 默认异步 compaction 也可能跟不上写入速度，出现 Log File 短时间堆积并不一定代表失败。

2.1.4 Append 表是不是可以不要 record key

不建议这么理解。Hudi 表仍然需要 record key。本文第一次创建 Append 表时没有定义 record key，实际报了 Primary key definition is required。所以即使写入操作是 insert，也要显式定义 Hudi record key；在 Flink SQL 中通常用 PRIMARY KEY NOT ENFORCED 声明，但它不是数据库强制唯一约束。

2.1.5 Bucket 数能不能后期调整

不建议随意调整。Bucket Index 的定位逻辑依赖固定 bucket 数，后期改 bucket 数会影响已有 key 到 FileGroup 的映射。如果业务增长很快，要在建表前认真估算 bucket 数。

2.2 高级能力使用边界

Flink Hudi 的这些高级能力，本质都不是“炫技参数”。

Changelog 是为了解决变化语义，Append 是为了避免不必要的索引开销，Clustering 是为了整理实时写入造成的文件布局问题，Bucket Index 是为了降低大表索引状态成本，Catalog 是为了让多表管理更像数据库。

我判断是否使用这些能力时，会先问几个问题：

1. 数据是否来自 CDC 链路？如果是，优先考虑 upsert 和 changelog 语义。
2. 数据是否天然只追加、不更新？如果是，优先考虑 insert / append。
3. 小文件是否已经影响查询？如果是，考虑 clustering。
4. state 是否已经影响 checkpoint？如果是，评估 Bucket Index。
5. 表数量是否开始变多？如果是，考虑 Hudi Catalog。

这些能力不要一次全开。先把基础写入、checkpoint、查询跑稳，再按问题引入对应能力。

3. 总结

本文整理了 Flink Hudi 的几个高级能力：

1. Changelog 模式用于保留和读取变化语义，但不能完全替代 CDC 原始日志。
2. Append 模式适合纯追加数据，但 Hudi 表仍然需要 record key。
3. Clustering 用于整理小文件和优化文件布局，生产实时链路更适合异步执行。
4. Bucket Index 通过固定 hash 桶降低索引定位成本，但 bucket 数不宜随意变更。
5. Hudi Catalog 可以用 DFS 或 HMS 模式统一管理表元数据。
6. 常见问题大多和 checkpoint、Hudi record key / 业务主键、precombine、compaction、slot 和索引策略有关。