Apache Hudi 学习笔记 11：Flink Hudi 核心参数与性能调优

上一篇已经用 Flink SQL 跑通了 Hudi COW 表写入和查询，也踩了一遍真实环境里的依赖冲突、checkpoint、slot 不够等问题。能跑通只是第一步，真正使用 Flink 写 Hudi 时，后面一定会遇到调优问题：写入跟不上 Kafka、checkpoint 变慢、状态越来越大、小文件越来越多、MOR 表 log file 变多、Compaction 抢资源。

本文专门整理 Flink Hudi 的核心参数。重点不只是列参数，而是把它们放回到 Flink Hudi 的写入链路里看：哪些参数影响去重，哪些影响并行度，哪些影响文件大小，哪些影响内存，哪些只在 MOR 表上有意义。

1. 参数模型基础

1.1 背景问题

Flink 写 Hudi 的调优经常会被简化成一句话：把 write.tasks 调大一点。这个方向有时有用，但不完整。

如果写入慢，可能有很多原因：

1. source 分区少，后面 writer 再多也没有数据。
2. write.bucket_assign.tasks 太少，key 到 FileGroup 的分配成为瓶颈。
3. write.tasks 太多，单个 TaskManager 内存不够，反而频繁 GC。
4. checkpoint 间隔太短，提交开销占比过高。
5. 分区太细，每个分区都产生小文件。
6. COW 表更新频繁，Base File 重写成本太高。
7. MOR 表 log file 太多，Compaction 没跟上。
8. RocksDB state 过大，checkpoint 变慢。

所以我更愿意把 Flink Hudi 调优理解成下面几个平衡：

调优方向	目标	代价
增加写入并行度	提高吞吐	更多 task、更多内存、更多小文件风险
增加 bucket assign 并行度	降低 key 分配瓶颈	state 分散，checkpoint 压力增加
增大写入 buffer	减少 flush、提升吞吐	单 task 内存压力增加
增大文件大小	减少小文件	单次写入和合并成本上升
增强 Compaction	降低 MOR 查询延迟	占用 CPU、IO 和内存
使用 RocksDB state	控制 JVM 内存	checkpoint 和本地磁盘 IO 需要关注

1.2 Flink Hudi 写入任务结构

先看写入链路。Flink Hudi 不是一个单一 sink，它内部包含 key 处理、bucket 分配、writer、checkpoint commit 等多个环节。

这张图主要说明 Flink Hudi 写入链路不是单个 sink，而是由 key 合并、bucket assigner、writer、checkpoint coordinator 和 Hudi commit 多个环节组成。

几个参数基本可以对应到图里的位置：

位置	关键参数
去重和合并	PRIMARY KEY、precombine.field
bucket 分配	write.bucket_assign.tasks
实际写文件	write.tasks、write.task.max.size、write.batch.size
state 初始化	write.index_bootstrap.tasks
查询读取	read.tasks
MOR 压缩	compaction.tasks、compaction.max_memory
checkpoint	execution.checkpointing.interval、execution.checkpointing.dir、state.backend.type

如果不知道某个参数该不该调，先问自己：它影响的是 source、bucket assigner、writer、compaction，还是 checkpoint？位置想清楚以后，调参会少很多盲目性。

1.3 去重参数：primary key 和 precombine.field

Flink SQL 建 Hudi 表时，一般用 PRIMARY KEY NOT ENFORCED 声明 Hudi record key / 业务主键字段：

CREATE TABLE demo_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_orders',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts'
);

这里的 NOT ENFORCED 是 Flink SQL 的语义：Flink 不会像数据库一样帮你强制检查唯一性；真正的更新合并还要结合 Hudi 写入语义、索引、precombine.field 和 checkpoint 提交结果。

precombine.field 决定同一个 key 多条记录如何合并。比如同一个 order_id=11002 有两条记录：

11002, CREATED, ts=1758074340000
11002, PAID,    ts=1758074520000

在本文 Hudi 1.0.2 Flink SQL upsert 示例中，precombine.field='ts' 用于同一 record key 多条候选记录的排序合并，ts 更大的记录会成为最终可见版本，所以最终状态应该是 PAID。如果使用不同版本、不同 merge mode 或自定义合并逻辑，需要以当前版本实测为准。

多字段 Hudi record key 可以这样写：

CREATE TABLE demo_order_detail (
  order_id BIGINT,
  item_id BIGINT,
  sku_id STRING,
  amount DECIMAL(10,2),
  ts BIGINT,
  dt STRING,
  PRIMARY KEY (order_id, item_id) NOT ENFORCED
) PARTITIONED BY (dt)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs:///hudi/demo_order_detail',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts'
);

我的经验是：precombine.field 必须选业务上能代表“新旧顺序”的字段。不要随便用处理时间 / 写入时间，也不要用可能回退的更新时间。如果使用处理时间作为 precombine 字段，重放历史数据、任务补偿、乱序事件和跨任务重试时，可能会把旧业务状态错误地覆盖成新版本。CDC 场景里更推荐使用 binlog 位点、事件时间加版本号，或者业务上严格单调的更新时间。

1.4 并行度参数

Flink Hudi 常见并行度参数如下：

参数	作用
write.tasks	实际写 Hudi 文件的并行度
write.bucket_assign.tasks	bucket assigner 并行度
write.index_bootstrap.tasks	全量接增量时加载已有索引的并行度
read.tasks	读取 Hudi 表的并行度
compaction.tasks	MOR compaction 并行度

这些参数不是越大越好。上一篇我在单机环境中就遇到过一个很典型的问题：write.tasks=2、write.bucket_assign.tasks=2 时，Flink Web UI 里 job 是 RUNNING，但顶点一直卡在 CREATED / SCHEDULED，因为可用 slot 不够。

单机测试可以先用下面这种保守配置：

WITH (
  'connector' = 'hudi',
  'path' = 'hdfs:///hudi/demo_flink_hudi_tuning_orders',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts',
  'write.tasks' = '1',
  'write.bucket_assign.tasks' = '1',
  'write.index_bootstrap.tasks' = '1',
  'read.tasks' = '1'
)

COW 表不依赖 MOR compaction，所以这个入门配置里不放 compaction.tasks。如果 Kafka 分区数、上游算子并行度或 shuffle key 分布本身有限，单纯调大 write.tasks 不会线性提高吞吐，反而可能制造更多小文件和 checkpoint 压力。

生产环境再按下面顺序放大：

1. 先看 source 并行度，比如 Kafka 分区数。
2. 再看 write.bucket_assign.tasks 是否成为瓶颈。
3. 再看 write.tasks 是否能吃满写入吞吐。
4. 最后看 checkpoint 时间、文件数量和 TaskManager 内存。

我不建议一开始就把所有并行度设置成同一个大数。bucket assigner、writer、compaction 的资源消耗不一样，应该分开调。

1.5 参数设置方式：WITH 和 SQL Hints

最常见的是在 DDL 的 WITH 中配置：

CREATE TABLE demo_orders (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_id STRING,
  ts BIGINT,
  dt STRING
) PARTITIONED BY (dt)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs:///hudi/demo_orders',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts',
  'write.tasks' = '4',
  'write.bucket_assign.tasks' = '2',
  'read.tasks' = '4'
);

如果只是临时调整某次读取或写入，也可以使用 Flink SQL 的 OPTIONS hint。比如临时调整读取并行度：

SET 'table.dynamic-table-options.enabled' = 'true';

SELECT *
FROM demo_orders /*+ OPTIONS('read.tasks'='8') */
WHERE dt = '2025-09-17';

或者临时覆盖写入参数：

INSERT INTO demo_orders /*+ OPTIONS('write.operation'='insert') */
SELECT order_id, user_id, ts, dt
FROM tmp_orders;

OPTIONS hint 是否能覆盖某个 Hudi 写入参数，需要以 Flink 1.20.1 + Hudi 1.0.2 当前 connector 实测为准；本文主线稳定参数仍建议写在 DDL WITH 中。也不要在线上任务中随意用 hint 把 upsert 临时改成 insert，只有确认输入全是新 record key、且上游已经去重时才适合这样做。

如果一次写入需要临时覆盖多个参数，可以写在同一个 OPTIONS hint 中：

INSERT INTO demo_orders
/*+ OPTIONS(
  'write.tasks'='<按 slot 和吞吐设置>',
  'write.bucket_assign.tasks'='<按 key 分配瓶颈设置>',
  'compaction.tasks'='<仅 MOR 表按 compaction 资源设置>'
) */
SELECT order_id, user_id, ts, dt
FROM tmp_orders;

我更倾向于把稳定参数放在 DDL 中，把临时调试参数放在 SQL Hints 中。否则线上表的长期行为会变得不容易追踪。OPTIONS hint 更适合压测、临时修复、或者同一张表在不同 SQL 中需要不同读取并行度的场景。

2. 核心调优参数

2.1 Compaction 参数

Compaction 只对 MOR 表有意义。COW 表更新时会重写 Base File，不存在 Base File + Log File 长期合并的问题；MOR 表为了降低写入延迟，会把更新追加到 Log File，后面再通过 Compaction 合并成新的 Base File。

这张图主要说明 MOR 表中 compaction 调度、执行、并行度和内存参数分别影响不同环节，不应只调一个 compaction.tasks。

常见 Compaction 参数：

参数	说明
compaction.schedule.enabled	是否调度 compaction plan
compaction.async.enabled	是否异步执行 compaction
compaction.trigger.strategy	触发策略，比如按 delta commit 数或时间
compaction.delta_commits	达到多少个 delta commit 触发
compaction.delta_seconds	达到多少秒触发
compaction.tasks	compaction 并行度
compaction.max_memory	单个 compaction task 可用内存，单位按 Hudi Flink 配置文档理解为 MB
compaction.target_io	单次 compaction 的目标 IO 预算

下面参数以 Hudi 1.0.2 Flink connector 配置为准，不要把 Spark DataSource 的 hoodie.compact.* 参数直接搬到 Flink SQL。官方 Flink Tuning Guide 中 compaction.trigger.strategy 的取值包括 num_commits、time_elapsed、num_and_time、num_or_time，本文使用的小写写法与 1.0.2 文档一致。

这里要把两个开关分开看：

参数组合	含义	适合场景
compaction.schedule.enabled=true、compaction.async.enabled=true	写入作业调度并异步执行 compaction	单作业实时写入，测试环境最容易观察
compaction.schedule.enabled=true、compaction.async.enabled=false	写入链路不在线执行 compaction；是否只生成 plan、何时由独立作业执行，需要结合部署方式验证	生产上想把压缩交给独立作业，避免和写入抢资源
compaction.schedule.enabled=false	不生成 compaction plan	只适合非常短期的测试，不适合长期 MOR 表

也就是说，compaction.async.enabled=false 不是“完全不需要 compaction”，而是把 compaction 执行从在线写入链路里拿出去。如果后面没有独立 compaction 作业，MOR 表的 Log File 会持续累积。

下面是 MOR 参数形态示例，不是本文第 10 节的主线实测表；真实 MOR 表还需要单独验证 log file、compaction plan、completed compaction 和查询结果。

CREATE TABLE demo_orders_mor (
  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_orders_mor',
  'table.type' = 'MERGE_ON_READ',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts',
  'compaction.schedule.enabled' = 'true',
  'compaction.async.enabled' = 'true',
  'compaction.trigger.strategy' = 'num_commits',
  'compaction.delta_commits' = '5',
  'compaction.tasks' = '2',
  'compaction.max_memory' = '512'
);

调 Compaction 的核心是平衡读写：

1. Compaction 太少：Log File 变多，Snapshot Query 合并成本上升。
2. Compaction 太频繁：写入链路和压缩任务抢 CPU / IO，checkpoint 也可能变慢。
3. Compaction 并行度太高：短期吞吐可能上去，但 TaskManager 内存压力变大。
4. compaction.max_memory 太小：合并 log 时容易 spill 或变慢。

2.2 文件大小参数

文件大小直接影响查询性能和元数据压力。小文件太多时，Hive、Spark、Flink、Trino 都会受到影响；文件太大时，更新和 Compaction 成本又会上升。

常见参数：

参数	说明
hoodie.parquet.max.file.size	Parquet 目标最大文件大小
hoodie.parquet.small.file.limit	小文件阈值
hoodie.copyonwrite.record.size.estimate	COW 表估算单条记录大小
hoodie.logfile.max.size	MOR Log File 最大大小

这些 hoodie.* 配置属于 Hudi 底层配置，Flink SQL WITH 中是否直接透传生效要以当前 Hudi 1.0.2 Flink connector 行为为准。本文第 10 节会通过 hoodie.properties、Timeline 和文件结果做基础验证，但运行时参数不一定都会持久化到表属性里。

这里有一个细节：Log File 的写入大小相对容易精确控制，但 Parquet 文件大小更多是按记录大小估算出来的。字段类型、压缩率、null 值比例、更新合并都会影响最终文件大小，所以 hoodie.parquet.max.file.size 更像目标值，不是严格上限。hoodie.copyonwrite.record.size.estimate 配得离真实记录大小太远时，COW 表 small file handling 的判断也会偏。

本文测试环境里使用了较小的值，方便观察：

WITH (
  'hoodie.parquet.max.file.size' = '67108864',
  'hoodie.parquet.small.file.limit' = '33554432',
  'hoodie.copyonwrite.record.size.estimate' = '512',
  'hoodie.logfile.max.size' = '134217728'
)

生产里常见 Parquet 目标大小会比这个大，比如 128MB、256MB。具体值要结合查询引擎、分区粒度、写入批次大小和 HDFS / S3 文件打开成本一起看。

我的建议是先观察三个指标：

1. 每个分区每天新增多少文件。
2. 单个 Parquet 文件平均大小。
3. 每次 upsert 是否频繁命中 small file handling。

如果每批数据很小，又按小时、城市、业务线继续细分分区，很容易每天产生大量小文件。这个时候比起盲目调大 write.tasks，更应该先检查分区设计和 batch 大小。

2.3 Hadoop per-job 参数

Flink 作业一般通过 HADOOP_CONF_DIR 读取 core-site.xml、hdfs-site.xml 等配置。但有些场景会希望把少量 Hadoop 参数写在表 DDL 中，比如同一个 Flink 集群临时写不同 HDFS 集群，或者某个作业需要单独设置 HDFS 客户端行为。

Hudi Flink connector 会读取 hadoop. 前缀的表选项，并去掉这个前缀后写入 Hadoop Configuration。比如：

CREATE TABLE demo_orders_remote_hdfs (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_id STRING,
  ts BIGINT,
  dt STRING
) PARTITIONED BY (dt)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs://hadoop1:8020/hudi/demo_orders_remote_hdfs',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts',
  'hadoop.fs.defaultFS' = 'hdfs://hadoop1:8020',
  'hadoop.dfs.replication' = '1',
  'hadoop.dfs.client.use.datanode.hostname' = 'true'
);

这里的 hadoop1:8020 只是远程 HDFS 示例地址，不是本文单机实测路径；实际使用时应替换成 hdfs://<namenode-host>:<port>。我不建议把大量 Hadoop 配置都塞进 DDL。稳定环境还是应该优先使用统一的 HADOOP_CONF_DIR，per-job 参数只适合补充少量和当前作业强相关的配置。也不要把访问密钥、真实账号密码这类敏感信息写进博客或长期提交到仓库，因为这些 DDL 选项可能进入 Catalog、作业日志或 SQL 审计记录。

2.4 内存参数

Flink Hudi 内存调优要放在 TaskManager 视角看。一个 TaskManager 里可能同时跑 source、bucket assigner、writer、compaction、checkpoint、network buffer、RocksDB。

这张图主要说明 TaskManager 内存需要同时容纳 writer buffer、bucket assigner、RocksDB state、network buffer、compaction 和 JVM 对象开销。

常见参数：

参数	说明
write.task.max.size	每个 write task 的最大写入 buffer，单位 MB
write.batch.size	写入批次大小，单位 MB
write.log_block.size	MOR log block 大小，单位按 Hudi Flink 配置文档理解为 MB
write.merge.max_memory	COW merge 可用内存，单位 MB
compaction.max_memory	compaction 合并 log 可用内存，单位 MB

COW 表更关注：

WITH (
  'write.task.max.size' = '1024',
  'write.batch.size' = '128',
  'write.merge.max_memory' = '512'
)

MOR 表除了 writer，还要关注 log 和 compaction：

WITH (
  'write.task.max.size' = '1024',
  'write.batch.size' = '128',
  'write.log_block.size' = '128',
  'compaction.max_memory' = '1024',
  'compaction.tasks' = '2'
)

这里不要把几个参数理解成严格扣减公式。更合理的检查方式是：

COW 场景：
  write.merge.max_memory 不应接近或超过 write.task.max.size

MOR 场景：
  compaction.max_memory、write.task.max.size、write.batch.size
  必须和 TaskManager 总内存、并行度一起评估

这能帮我快速识别明显不合理的配置。比如 write.task.max.size=1024，却给 write.merge.max_memory=1024，那普通写入 buffer、网络 buffer、对象开销都没有空间，实际运行时很容易变成 GC 或 spill 问题。write.batch.size 也不应长期大于或接近 write.task.max.size，否则单批数据就可能挤压 merge buffer、network buffer 和对象开销空间。

这里最容易犯的错误是只看单个参数，不看 TaskManager 总内存。比如 TaskManager 只有 4GB，却同时跑 2 个 writer、2 个 bucket assigner、1 个 compaction task，还给每个 writer 配 2GB buffer，这种配置很容易把 JVM 和 RocksDB 一起压垮。

3. 实测验证

3.1 实战：写入一张调优参数测试表

下面用一张 COW 表验证本文涉及的部分参数。测试数据很小，只用来确认 DDL、写入、commit、查询链路。

准备 SQL。这里的 dt 是模拟业务分区日期，用于演示多分区写入和文件布局，不代表文章实际写作日期；ts 是模拟业务事件时间，用作 precombine.field 排序字段：

cat > /tmp/hudi_flink_sql_tuning_demo.sql <<'SQL'
SET 'execution.runtime-mode' = 'streaming';
SET 'sql-client.execution.result-mode' = 'TABLEAU';
SET 'execution.checkpointing.interval' = '5s';
SET 'execution.checkpointing.dir' = 'hdfs:///flink-checkpoints/hudi_sql_tuning_demo';
SET 'state.backend.type' = 'rocksdb';

DROP TABLE IF EXISTS demo_flink_hudi_tuning_orders;

CREATE TABLE demo_flink_hudi_tuning_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_flink_hudi_tuning_orders',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts',
  'hoodie.datasource.write.recordkey.field' = 'order_id',
  'hoodie.datasource.write.partitionpath.field' = 'dt',
  'hoodie.metadata.enable' = 'false',
  'write.tasks' = '1',
  'write.bucket_assign.tasks' = '1',
  'write.index_bootstrap.tasks' = '1',
  'read.tasks' = '1',
  'write.task.max.size' = '512',
  'write.batch.size' = '64',
  'write.merge.max_memory' = '128',
  'hoodie.parquet.max.file.size' = '67108864',
  'hoodie.parquet.small.file.limit' = '33554432',
  'hoodie.copyonwrite.record.size.estimate' = '512'
);

INSERT INTO demo_flink_hudi_tuning_orders VALUES
  (11001, 'u101', CAST(19.90 AS DECIMAL(10,2)), 'CREATED', 1758074280000, '2025-09-17'),
  (11002, 'u102', CAST(88.00 AS DECIMAL(10,2)), 'CREATED', 1758074340000, '2025-09-17'),
  (11003, 'u103', CAST(66.60 AS DECIMAL(10,2)), 'CREATED', 1758074400000, '2025-09-18'),
  (11002, 'u102', CAST(88.00 AS DECIMAL(10,2)), 'PAID', 1758074520000, '2025-09-17'),
  (11004, 'u104', CAST(128.80 AS DECIMAL(10,2)), 'CREATED', 1758074580000, '2025-09-18');
SQL

本文显式写出 DataSource 风格的 record key / partition path 配置，是为了让 Hudi 表属性更清晰；在 Flink SQL 中，部分信息也可以从 PRIMARY KEY NOT ENFORCED 和 PARTITIONED BY 推断，具体以 Hudi 1.0.2 Flink connector 行为为准。hoodie.metadata.enable=false 只是为了减少单机测试变量，生产环境是否开启 Metadata Table 要结合表规模、文件数量、查询方式和 Hudi/Flink connector 版本评估。

执行：

unset HADOOP_CLASSPATH
export HADOOP_CONF_DIR=/opt/module/hadoop-3.4.1/etc/hadoop

hdfs dfs -rm -r -f \
  /hudi/demo_flink_hudi_tuning_orders \
  /flink-checkpoints/hudi_sql_tuning_demo

/opt/module/flink-1.20.1/bin/sql-client.sh \
  -f /tmp/hudi_flink_sql_tuning_demo.sql

本文 Hadoop 安装目录按 /opt/module/hadoop-3.4.1 记录；如果你的机器使用 /opt/hadoop-3.4.1 或 /opt/hadoop 软链接，按实际路径替换即可。这里 unset HADOOP_CLASSPATH 是为了避免把本机 Hadoop 依赖再次通过环境变量注入 Flink，造成和 Flink lib 中已有依赖重复。本文依赖主要放在 Flink lib 中，再通过 HADOOP_CONF_DIR 提供 Hadoop 配置。

手工删除 HDFS 表路径和 checkpoint 目录只是为了保证单机测试可重复执行；生产环境不要直接 hdfs dfs -rm Hudi 表目录或 checkpoint 目录，应按表治理、回滚、重建和 savepoint / checkpoint 规范处理。本文 Flink 1.20.1 环境中 state.backend.type=rocksdb 实测可用；如果当前发行包要求 embedded-rocksdb 或其他配置项，应按 Flink 1.20 文档调整。

SQL Client 输出，下面只截取 CREATE TABLE 和 INSERT 的关键输出：

Flink SQL>
> CREATE TABLE demo_flink_hudi_tuning_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_flink_hudi_tuning_orders',
>   'table.type' = 'COPY_ON_WRITE',
>   'write.operation' = 'upsert',
>   'precombine.field' = 'ts',
>   'hoodie.datasource.write.recordkey.field' = 'order_id',
>   'hoodie.datasource.write.partitionpath.field' = 'dt',
>   'hoodie.metadata.enable' = 'false',
>   'write.tasks' = '1',
>   'write.bucket_assign.tasks' = '1',
>   'write.index_bootstrap.tasks' = '1',
>   'read.tasks' = '1',
>   'write.task.max.size' = '512',
>   'write.batch.size' = '64',
>   'write.merge.max_memory' = '128',
>   'hoodie.parquet.max.file.size' = '67108864',
>   'hoodie.parquet.small.file.limit' = '33554432',
>   'hoodie.copyonwrite.record.size.estimate' = '512'
> )
[INFO] Execute statement succeeded.

Flink SQL>
> INSERT INTO demo_flink_hudi_tuning_orders VALUES
>   (11001, 'u101', CAST(19.90 AS DECIMAL(10,2)), 'CREATED', 1758074280000, '2025-09-17'),
>   (11002, 'u102', CAST(88.00 AS DECIMAL(10,2)), 'CREATED', 1758074340000, '2025-09-17'),
>   (11003, 'u103', CAST(66.60 AS DECIMAL(10,2)), 'CREATED', 1758074400000, '2025-09-18'),
>   (11002, 'u102', CAST(88.00 AS DECIMAL(10,2)), 'PAID', 1758074520000, '2025-09-17'),
>   (11004, 'u104', CAST(128.80 AS DECIMAL(10,2)), 'CREATED', 1758074580000, '2025-09-18')
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: d4132eb25e276e1da9237db21d30bbcf

虽然 runtime mode 是 streaming，但本文的 VALUES 输入是有界数据，写完后 job 会结束；如果 source 是 Kafka CDC 这类无界流，job 正常情况下会长期运行，checkpoint 是否持续成功才是关键。

3.2 验证 Timeline 和文件

查看 Timeline：

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

输出：

Found 4 items
-rw-r--r--   1 root supergroup          0 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/.hoodie/timeline/20250918095908714.commit.requested
-rw-r--r--   1 root supergroup          0 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/.hoodie/timeline/20250918095908714.inflight
-rw-r--r--   1 root supergroup       4400 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/.hoodie/timeline/20250918095908714_20250918095910660.commit
drwxr-xr-x   - root supergroup          0 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/.hoodie/timeline/history

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

查看 checkpoint 元信息和数据文件：

hdfs dfs -ls -R /hudi/demo_flink_hudi_tuning_orders \
  | grep -E 'ckp_meta|2025-09-1[89]|parquet'

输出：

drwxr-xr-x   - root supergroup          0 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/.hoodie/.aux/ckp_meta
-rw-r--r--   1 root supergroup          0 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/.hoodie/.aux/ckp_meta/20250918095908714.COMPLETED
-rw-r--r--   1 root supergroup          0 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/.hoodie/.aux/ckp_meta/20250918095908714.INFLIGHT
drwxr-xr-x   - root supergroup          0 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/2025-09-17
-rw-r--r--   1 root supergroup     435945 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/2025-09-17/b8443c8a-68e1-4e8a-82df-92c01d1c67e8_0-1-0_20250918095908714.parquet
drwxr-xr-x   - root supergroup          0 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/2025-09-18
-rw-r--r--   1 root supergroup     436015 2025-09-18 09:59 /hudi/demo_flink_hudi_tuning_orders/2025-09-18/6f16373b-ca9e-4228-891d-3cf7671dc967_0-1-0_20250918095908714.parquet

.hoodie/.aux/ckp_meta 中是 Flink Hudi 写入过程使用的 checkpoint / instant 辅助元信息。它和 Flink 的 checkpoint 存储目录、Kafka source checkpoint、HoodieStreamer checkpoint 不是同一个概念，不要混淆。

数据文件名前半段是 fileId，中间的 0-1-0 类似写入任务相关 token，最后的 20250918095908714 是写入 instant，不要把整个文件名都理解成 fileId。这里文件只有 436KB 左右，是因为本文只有 5 条小样本数据，每个分区数据量远小于 small file limit；文件大小参数更多在持续写入或大批量数据中体现效果。

查看表属性：

hdfs dfs -cat /hudi/demo_flink_hudi_tuning_orders/.hoodie/hoodie.properties \
  | grep -E 'hoodie.table.type|hoodie.table.name|hoodie.table.version'

输出：

hoodie.table.type=COPY_ON_WRITE
hoodie.table.name=demo_flink_hudi_tuning_orders
hoodie.table.version=8

hoodie.properties 主要保存表级元信息，不代表 DDL WITH 中所有运行参数都会完整写入这里。调运行时参数是否生效，还要结合 Flink job 配置、日志、Timeline、文件结果和 Web UI 指标判断。

3.3 查询验证

准备查询 SQL。查询脚本使用 batch runtime，只读取已经完成的 Hudi commit，不需要配置 checkpoint。这里不是重新写表，而是在 Flink SQL 中用同样 schema 注册已有 Hudi 路径用于查询；字段、PRIMARY KEY NOT ENFORCED、PARTITIONED BY、table.type 应和已有表配置保持一致。

cat > /tmp/hudi_flink_sql_tuning_query.sql <<'SQL'
SET 'execution.runtime-mode' = 'batch';
SET 'sql-client.execution.result-mode' = 'TABLEAU';

CREATE TABLE demo_flink_hudi_tuning_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_flink_hudi_tuning_orders',
  'table.type' = 'COPY_ON_WRITE',
  'hoodie.metadata.enable' = 'false',
  'read.tasks' = '1'
);

SELECT dt, COUNT(*) AS cnt, MIN(order_id) AS min_order_id, MAX(order_id) AS max_order_id
FROM demo_flink_hudi_tuning_orders
GROUP BY dt
ORDER BY dt;

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

执行：

/opt/module/flink-1.20.1/bin/sql-client.sh \
  -f /tmp/hudi_flink_sql_tuning_query.sql

分区统计：

+------------+-----+--------------+--------------+
|         dt | cnt | min_order_id | max_order_id |
+------------+-----+--------------+--------------+
| 2025-09-17 |   2 |        11001 |        11002 |
| 2025-09-18 |   2 |        11003 |        11004 |
+------------+-----+--------------+--------------+
2 rows in set (5.87 seconds)

明细结果：

+----------+---------+--------+---------+------------+
| order_id | user_id | amount |  status |         dt |
+----------+---------+--------+---------+------------+
|    11001 |    u101 |  19.90 | CREATED | 2025-09-17 |
|    11002 |    u102 |  88.00 |    PAID | 2025-09-17 |
|    11003 |    u103 |  66.60 | CREATED | 2025-09-18 |
|    11004 |    u104 | 128.80 | CREATED | 2025-09-18 |
+----------+---------+--------+---------+------------+
4 rows in set (1.20 seconds)

这里可以看到 order_id=11002 最终是 PAID，说明 Flink SQL 中声明的 PRIMARY KEY NOT ENFORCED，也就是 Hudi record key / 业务主键字段，配合 precombine.field=ts 的更新合并逻辑生效。

4. 常见问题与调优顺序

4.1 常见问题

4.1.1 write.tasks 调大后为什么反而更慢

常见原因是 slot、内存、checkpoint 或小文件压力上来了。write.tasks 增大以后，writer task 数变多，单个 TaskManager 同时运行的 task 也变多。如果 TaskManager 内存没有同步调整，很容易出现 GC、spill、checkpoint 变慢。

建议先看：

/opt/module/flink-1.20.1/bin/flink list

再看 Flink Web UI 中 task 是否一直处于 CREATED、SCHEDULED 或频繁重启。

4.1.2 为什么 checkpoint 成功前看不到 commit

Flink Hudi sink 依赖 checkpoint 推进 commit。数据写入临时文件或 buffer 不代表 Hudi commit 已经完成，只有 checkpoint 协调完成后，Timeline 中才会出现 completed commit。本文 VALUES 小样本是为了快速验证链路；真正的 Kafka CDC 长期流任务中，checkpoint 是否持续成功才是 Hudi commit 稳定推进的关键。

可以检查：

hdfs dfs -ls /hudi/demo_flink_hudi_tuning_orders/.hoodie/timeline
hdfs dfs -ls /hudi/demo_flink_hudi_tuning_orders/.hoodie/.aux/ckp_meta

4.1.3 COW 表需要配置 compaction 吗

不需要。Compaction 是 MOR 表的核心表服务。COW 表写入时已经重写 Base File，不需要后续把 Log File 合并成 Base File。

如果 COW 表查询变慢，更应该先看小文件、分区设计、写入批次和 clustering。

4.1.4 MOR 表只有 log 文件是不是异常

不一定。MOR 表更新写入时追加 Log File 是正常现象。需要判断 compaction 是否按预期触发，Snapshot Query 是否能读到最新结果，Read Optimized Query 是否只读到已压缩的 Base File。如果 Read Optimized 查询看不到最新更新，而 Snapshot Query 能看到，通常说明最新变更仍在 log file 中，尚未完成 compaction，这是 MOR 语义的一部分。

4.1.5 write.batch.size 和 write.task.max.size 怎么取值

这两个值要结合 TaskManager 内存和 writer 并行度。一个非常粗略的估算方式是：

单个 write task 可用内存
  < TaskManager 可用于用户算子的内存 / 同时运行的 writer task 数

这不是 Flink 的严格内存分配公式，真实可用内存还要扣除 managed memory、network memory、RocksDB、本地排序、checkpoint 和 JVM 对象开销。不要把 TaskManager 内存都分给 writer buffer。

4.1.6 为什么开启 Compaction 后还是没有新的 Parquet 文件

先分清是没有调度 plan，还是调度了但执行失败。下面命令针对第 6 节 MOR 示例表，不适用于本文第 10 节 COW 实测表；COW 表不会产生 compaction。

hdfs dfs -ls /hudi/demo_orders_mor/.hoodie/timeline \
  | grep compaction

如果只看到 compaction requested / inflight，没有 completed，常见原因有三个：

1. compaction.async.enabled=false，写入作业只负责生成 plan，没有在线执行。
2. TaskManager 缺 Hadoop 运行时依赖，compaction task 失败，但 SQL Client 只看到 job 异常。
3. compaction.tasks 或 compaction.max_memory 配得太激进，任务反复重启。

如果没有搜到 compaction，也要结合 MOR 表的 deltacommit、compaction plan 目录、Flink TaskManager 日志和当前 Hudi 1.0.2 Timeline 布局继续排查。

排查时不要只看 SQL Client，要去 TaskManager 日志里搜 compaction、ClassNotFoundException、NoSuchMethodError、FileSystem。如果是 Hadoop 依赖问题，优先确认 HADOOP_CONF_DIR、Flink lib 目录里的 Hudi bundle、以及是否误设置了会污染 classpath 的 HADOOP_CLASSPATH。

4.2 Flink Hudi 调优顺序

Flink Hudi 调优最重要的是先分清“瓶颈在哪一段”。如果 source 只有 2 个 Kafka 分区，把 write.tasks 调到 32 并不会让吞吐线性提升；如果 bucket assigner 是瓶颈，只调 writer 也没有用；如果小文件已经失控，继续增加 writer 并行度可能只是制造更多文件。

我更倾向于按这个顺序调：

1. 先跑通低并行度链路，确认 checkpoint 和 commit 正常。
2. 观察 source 吞吐、bucket assigner、writer、checkpoint 的耗时。
3. 逐步放大 write.bucket_assign.tasks 和 write.tasks。
4. 同时观察文件大小和文件数量。
5. 如果是 MOR 表，再单独调 compaction。
6. 最后才调内存上限和 GC。

调参最怕一次改十个参数。那样即使吞吐变好了，也不知道是哪一个参数起作用；如果变差了，也很难定位。更稳妥的做法是每次只改一到两个参数，并记录吞吐、checkpoint duration、文件数量、文件大小、TaskManager GC 和失败重启次数。

5. 总结

本文整理了 Flink Hudi 参数调优的核心思路：

1. PRIMARY KEY NOT ENFORCED 和 precombine.field 决定 Hudi record key 与更新顺序。
2. write.tasks、write.bucket_assign.tasks、read.tasks、compaction.tasks 要结合 slot 和实际瓶颈配置。
3. MOR 表才需要重点关注 Compaction，COW 表更关注文件大小和写放大。
4. 文件大小参数影响查询性能、小文件数量和更新成本。
5. Hadoop per-job 参数可以通过 hadoop. 前缀写入 DDL，但不要滥用。
6. 内存参数必须放到 TaskManager 总内存和并行度里一起计算。
7. Flink Hudi commit 依赖 checkpoint，排查时一定要看 Timeline 和 .hoodie/.aux/ckp_meta。