前面两篇已经把 Flink SQL 写 Hudi 和核心参数调优讲了一遍。本文继续往实时入湖方向走，重点看三个问题：Flink 怎么流式读取 Hudi 表？CDC 数据怎么从 Kafka 写入 Hudi？离线全量数据和实时增量数据怎么衔接？
这三个问题其实是一条链路上的不同阶段。CDC 入湖负责把变化写进去，Hudi Streaming Query 负责把新的 commit 持续读出来，全量接增量负责解决“实时任务启动前，表里已经有历史数据”的识别问题。
1. 读取和 CDC 链路
1.1 背景问题
实时入湖不是简单地把数据写到 HDFS。以 MySQL 订单表为例，真实链路里至少有这些问题：
- MySQL binlog 里既有 insert，也有 update 和 delete。
- 同一个 Hudi record key / 业务主键可能短时间内多次变化。
- Kafka 中的 CDC 消息需要能回放。
- Flink 任务失败后要从 checkpoint 恢复。
- Hudi 表需要根据 record key 做 upsert，而不是简单追加。
- 下游任务希望持续消费 Hudi 新 commit。
- 全量导入以后，实时任务不能把已有 record key 当成新记录重复插入。
我把这类链路拆成四层：
| 层次 | 作用 |
|---|---|
| CDC Source | 从 MySQL binlog 抽取变化 |
| Kafka | 保存可回放的 CDC 事件日志 |
| Flink SQL | 解析 CDC、维护状态、写入 Hudi |
| Hudi | 提供 upsert、Timeline、增量读取 |
本文实操部分用 Debezium JSON 消息模拟 MySQL binlog 已经进入 Kafka 的状态。这样可以把重点放在 Flink Hudi 的读写语义上，不把篇幅消耗在 MySQL CDC connector 的安装细节上。
1.2 Flink Hudi 读取方式
Flink 读 Hudi 常见有三类：
| 读取方式 | 说明 | 适合场景 |
|---|---|---|
| Snapshot Query | 读取当前最新快照 | 批量校验、临时查询、维表回填 |
| Streaming Query | 持续监听新 commit | 增量消费、实时下游 pipeline |
| Time Travel | 读取指定时间点附近快照 | 数据回溯、问题排查 |
Flink Streaming Read 的核心参数是：
'read.streaming.enabled' = 'true',
'read.start-commit' = 'earliest',
'read.streaming.check-interval' = '4'它的工作方式可以理解成下面这样：
这张图主要说明 Flink Hudi streaming source 会周期性轮询 Hudi Timeline，从指定 start commit 附近开始持续读取新的 completed instant。
read.start-commit 可以指定从哪个 instant 开始读。常见取值：
| 参数值 | 含义 |
|---|---|
earliest |
从最早可用 commit 开始消费 |
| 具体 instant | 从指定 instant 附近开始消费；是否包含该 instant 需要以 Hudi 1.0.2 Flink connector 实测语义为准 |
| 不设置 | 通常从最新位置开始监听后续 commit |
常见读取参数可以先记住这几个：
| 参数 | 说明 |
|---|---|
read.streaming.enabled |
是否开启流式读取 |
read.streaming.check-interval |
流读轮询 Timeline 的间隔，单位秒，默认 60 |
read.start-commit |
增量读取起始 instant，支持 earliest |
read.end-commit |
增量读取结束 instant |
read.streaming.skip_compaction |
流读时是否跳过 compaction instant，Hudi 1.0.2 默认开启 |
read.streaming.skip_clustering |
流读时是否跳过 clustering instant，Hudi 1.0.2 默认开启 |
流读要特别注意清理策略。如果下游流读任务落后太多，而 Hudi clean 已经删除了它需要的历史 FileSlice，就可能读不到旧 commit 对应的数据。所以流读链路里，clean.retain_commits 不能只按写入端考虑，还要给下游消费延迟留余量。
read.streaming.skip_compaction=true 可以减少流读重复扫描 compaction 产物的开销，但它不是万能开关。MOR 表如果下游落后超过 clean.retain_commits 能保留的范围，旧的 FileSlice 已经被清理，流读仍然可能丢失需要补读的数据。我的经验是：只要下游依赖 Hudi 流读做准实时 pipeline，就要把 clean 保留策略和下游最大允许落后时间一起设计。
1.3 实测：Flink Streaming Read 读取 Hudi 表
后面 CDC 入湖会生成一张表：hdfs:///hudi/demo_flink_cdc_orders。这一节是基于第 6、7 节已经写入完成后的验证结果，阅读时可以先看完 CDC 写入部分，再回来看流读 SQL：
cat > /tmp/hudi_flink_stream_read.sql <<'SQL'
SET 'execution.runtime-mode' = 'streaming';
SET 'sql-client.execution.result-mode' = 'TABLEAU';
CREATE TABLE hudi_orders_stream_read (
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_cdc_orders',
'table.type' = 'COPY_ON_WRITE',
'hoodie.metadata.enable' = 'false',
'read.tasks' = '1',
'read.streaming.enabled' = 'true',
'read.start-commit' = 'earliest',
'read.streaming.check-interval' = '4'
);
SELECT order_id, user_id, status, dt
FROM hudi_orders_stream_read
WHERE order_id IN (12001, 12002, 12003)
LIMIT 3;
SQL执行：
unset HADOOP_CLASSPATH
export HADOOP_CONF_DIR=/opt/module/hadoop-3.4.1/etc/hadoop
/opt/module/flink-1.20.1/bin/sql-client.sh \
-f /tmp/hudi_flink_stream_read.sql输出如下：
+----+----------------------+--------------------------------+--------------------------------+--------------------------------+
| op | order_id | user_id | status | dt |
+----+----------------------+--------------------------------+--------------------------------+--------------------------------+
| +I | 12002 | u202 | PAID | 2025-09-28 |
| +I | 12001 | u201 | CREATED | 2025-09-28 |
| +I | 12003 | u203 | CREATED | 2025-09-29 |
+----+----------------------+--------------------------------+--------------------------------+--------------------------------+
3 rows in set这个结果说明，在 read.start-commit='earliest' 的配置下，Flink Hudi streaming source 可以从已有 commit 开始读取历史数据；如果不设置 start commit，通常更像从当前最新位置监听后续 commit。因为这里加了 LIMIT 3，SQL Client 输出 3 条后结束；如果不加限制，它会持续等待后续 Hudi commit。
如果要观察持续监听，可以去掉 LIMIT：
SELECT order_id, user_id, status, dt
FROM hudi_orders_stream_read;如果只想做批式增量读取，可以同时指定 read.start-commit 和 read.end-commit。在本文 Hudi 1.0.2 Flink connector 实测中，指定 start 和 end commit 的 batch incremental read 会读取该范围内的变更数据；起止边界是否包含对应 instant，建议以当前版本文档和实测为准。如果要避免重复消费上一个 checkpoint，建议把 start commit 设置为上次已处理 completed instant 之后的下一个起点，或按当前版本文档确认边界语义。
CREATE TABLE hudi_orders_incremental_read (
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_cdc_orders',
'table.type' = 'COPY_ON_WRITE',
'hoodie.metadata.enable' = 'false',
'read.start-commit' = '20250929203945218',
'read.end-commit' = '20250929203949912'
);
SELECT order_id, status, dt
FROM hudi_orders_incremental_read
ORDER BY order_id;在 Flink Hudi 中，可以通过限定 read.end-commit 来读取某个 instant 范围内的数据；如果要严格表达 Time Travel 快照语义，需要按 Hudi 1.0.2 Flink 查询文档和实测确认：
CREATE TABLE hudi_orders_time_travel (
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_cdc_orders',
'table.type' = 'COPY_ON_WRITE',
'hoodie.metadata.enable' = 'false',
'read.end-commit' = '20250929203949912'
);1.4 CDC 入湖整体链路
CDC 入湖可以有两种常见做法：
- Flink CDC connector 直接读取 MySQL binlog，然后写 Hudi。
- Debezium / Canal / Flink CDC 先把 binlog 写入 Kafka，再由 Flink SQL 消费 Kafka 写 Hudi。
我更喜欢第二种，尤其是生产环境。原因是 Kafka 能保存 CDC 原始事件，方便回放、补偿、排查，也能把数据库采集和湖表写入解耦。
本文实测链路如下：
这张图主要说明 MySQL binlog 先进入 Kafka，再由 Flink SQL 解析 Debezium JSON changelog，最终通过 Hudi upsert 写入湖表。
如果要把 MySQL binlog 先写入 Kafka，Flink SQL 的形态一般是 mysql-cdc 源表加 upsert-kafka 目标表。下面是结构示例，<mysql-host> 替换为真实 MySQL 地址，账号和口令在博客里统一使用测试值：
CREATE TABLE mysql_orders_cdc (
order_id BIGINT,
user_id STRING,
amount DECIMAL(10,2),
status STRING,
ts BIGINT,
dt STRING,
PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = '<mysql-host>',
'port' = '3306',
'username' = 'test_user',
'password' = 'test_password',
'database-name' = 'test',
'table-name' = 'orders'
);
CREATE TABLE kafka_orders_upsert (
order_id BIGINT,
user_id STRING,
amount DECIMAL(10,2),
status STRING,
ts BIGINT,
dt STRING,
PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'hudi_flink_cdc_orders_upsert',
'properties.bootstrap.servers' = 'localhost:9092',
'key.format' = 'json',
'value.format' = 'json'
);
INSERT INTO kafka_orders_upsert
SELECT order_id, user_id, amount, status, ts, dt
FROM mysql_orders_cdc;本文账号和口令都只用于本地单机示例；生产环境应使用最小权限账号、强密码，并通过安全配置管理，不要把真实密码写入博客或仓库。
这条链路适合把数据库采集和入湖写入拆开。需要注意，upsert-kafka 表输出的是按 key 更新后的 changelog，不是 Debezium 原始 JSON 的 before / after / op 结构。本文后面的实测从 Kafka Debezium JSON 开始，假设 Kafka 中已经是 Debezium JSON 格式，是为了把重点放在 Hudi 写入和读取语义上。
需要补一个 Flink Kafka SQL connector：
cd /opt/module/flink-1.20.1/lib
curl -k -L \
-o flink-sql-connector-kafka-3.4.0-1.20.jar \
https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-kafka/3.4.0-1.20/flink-sql-connector-kafka-3.4.0-1.20.jar验证 jar 中有 Kafka SQL connector：
jar tf flink-sql-connector-kafka-3.4.0-1.20.jar \
| grep KafkaDynamicTableFactory输出：
org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.class修改 Flink lib 后重启 Flink：
/opt/module/flink-1.20.1/bin/stop-cluster.sh
/opt/module/flink-1.20.1/bin/start-cluster.sh如果测试机系统时间早于证书生效时间，远端 Maven 证书校验可能失败。正常环境应恢复正确系统时间、开启 NTP，并使用 curl -L 或 wget 保留证书校验；curl -k 只适合隔离测试环境临时排错。本文使用的 flink-sql-connector-kafka-3.4.0-1.20.jar 与 Flink 1.20 配套，实际版本仍以对应版本的官方 connector 兼容说明为准。
2. CDC 写入验证
2.1 准备 Kafka CDC Topic
创建 Topic：
/opt/module/kafka/bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--create \
--if-not-exists \
--topic hudi_flink_cdc_orders \
--partitions 1 \
--replication-factor 1如果 Topic 不存在，输出类似：
Created topic hudi_flink_cdc_orders.准备 Debezium JSON 消息。这里的 dt 是模拟业务分区日期，用于演示多分区 CDC 入湖，不代表文章实际写作日期；ts 是模拟业务事件时间，用作 Hudi precombine.field 排序字段：
cat > /tmp/hudi_flink_cdc_orders.jsonl <<'JSONL'
{"before":null,"after":{"order_id":12001,"user_id":"u201","amount":39.90,"status":"CREATED","ts":1759063600000,"dt":"2025-09-28"},"op":"c","ts_ms":1759063601000}
{"before":null,"after":{"order_id":12002,"user_id":"u202","amount":88.00,"status":"CREATED","ts":1759063660000,"dt":"2025-09-28"},"op":"c","ts_ms":1759063661000}
{"before":null,"after":{"order_id":12003,"user_id":"u203","amount":168.00,"status":"CREATED","ts":1759063720000,"dt":"2025-09-29"},"op":"c","ts_ms":1759063721000}
{"before":{"order_id":12002,"user_id":"u202","amount":88.00,"status":"CREATED","ts":1759063660000,"dt":"2025-09-28"},"after":{"order_id":12002,"user_id":"u202","amount":88.00,"status":"PAID","ts":1759063900000,"dt":"2025-09-28"},"op":"u","ts_ms":1759063901000}
JSONL写入 Kafka：
/opt/module/kafka/bin/kafka-console-producer.sh \
--bootstrap-server localhost:9092 \
--topic hudi_flink_cdc_orders \
< /tmp/hudi_flink_cdc_orders.jsonl消费验证：
/opt/module/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic hudi_flink_cdc_orders \
--from-beginning \
--max-messages 4 \
--timeout-ms 10000输出：
{"before":null,"after":{"order_id":12001,"user_id":"u201","amount":39.90,"status":"CREATED","ts":1759063600000,"dt":"2025-09-28"},"op":"c","ts_ms":1759063601000}
{"before":null,"after":{"order_id":12002,"user_id":"u202","amount":88.00,"status":"CREATED","ts":1759063660000,"dt":"2025-09-28"},"op":"c","ts_ms":1759063661000}
{"before":null,"after":{"order_id":12003,"user_id":"u203","amount":168.00,"status":"CREATED","ts":1759063720000,"dt":"2025-09-29"},"op":"c","ts_ms":1759063721000}
{"before":{"order_id":12002,"user_id":"u202","amount":88.00,"status":"CREATED","ts":1759063660000,"dt":"2025-09-28"},"after":{"order_id":12002,"user_id":"u202","amount":88.00,"status":"PAID","ts":1759063900000,"dt":"2025-09-28"},"op":"u","ts_ms":1759063901000}
Processed a total of 4 messages本文实测只演示 Debezium insert / update 进入 Hudi；delete 事件需要单独验证，尤其要确认 Flink debezium-json 如何把 delete changelog 转给 Hudi sink。
2.2 Flink SQL：Kafka CDC 写入 Hudi
准备 SQL：
cat > /tmp/hudi_flink_cdc_kafka_to_hudi.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_flink_cdc_orders';
SET 'state.backend.type' = 'rocksdb';
SET 'parallelism.default' = '1';
DROP TABLE IF EXISTS kafka_orders_cdc;
DROP TABLE IF EXISTS hudi_orders_cdc;
CREATE TABLE kafka_orders_cdc (
order_id BIGINT,
user_id STRING,
amount DECIMAL(10,2),
status STRING,
ts BIGINT,
dt STRING,
PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
'connector' = 'kafka',
'topic' = 'hudi_flink_cdc_orders',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'hudi-flink-cdc-orders-g20250929',
'scan.startup.mode' = 'earliest-offset',
'format' = 'debezium-json'
);
CREATE TABLE hudi_orders_cdc (
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_cdc_orders',
'table.type' = 'COPY_ON_WRITE',
'write.operation' = 'upsert',
'precombine.field' = 'ts',
'hoodie.metadata.enable' = 'false',
'write.tasks' = '1',
'write.bucket_assign.tasks' = '1',
'read.tasks' = '1',
'write.rate.limit' = '1000'
);
INSERT INTO hudi_orders_cdc
SELECT order_id, user_id, amount, status, ts, dt
FROM kafka_orders_cdc;
SQL这里有几个关键点：
- Kafka 表使用
format='debezium-json'。 - Hudi 表使用
write.operation='upsert'。 precombine.field='ts'用于同一 record key 多版本合并。write.rate.limit='1000'用于限制写入速率。- 单机环境只有 1 个 slot，所以并行度都设为 1。
kafka_orders_cdc 的 primary key 主要用于表达 changelog key 语义；不同 Flink/Kafka connector 配置下，可能还需要 key.format、key.fields 或 upsert-kafka 方案。如果复用旧 group.id，可能不会重新消费已有 Kafka 消息；测试时建议使用新的 Topic 或新的 group.id。本文 Flink 1.20.1 环境中 state.backend.type=rocksdb 实测可用，如果当前发行包要求 embedded-rocksdb 或其他配置项，应按 Flink 1.20 文档调整。
write.rate.limit 在 Hudi 1.0.2 Flink connector 中用于限制每秒写入记录数，本文用 1000 只是单机测试值。当前 CDC 实测是从空 Hudi 表开始写入，所以没有开启 index.bootstrap.enabled；如果目标表已经有历史全量数据，应参考第 11 节开启 index bootstrap 或使用已有 checkpoint / savepoint state。
执行：
unset HADOOP_CLASSPATH
export HADOOP_CONF_DIR=/opt/module/hadoop-3.4.1/etc/hadoop
hdfs dfs -rm -r -f \
/hudi/demo_flink_cdc_orders \
/flink-checkpoints/hudi_flink_cdc_orders
/opt/module/flink-1.20.1/bin/sql-client.sh \
-f /tmp/hudi_flink_cdc_kafka_to_hudi.sql这里 unset HADOOP_CLASSPATH 是为了避免把本机 Hadoop 依赖再次通过环境变量注入 Flink，造成和 Flink lib 中已有依赖重复。本文依赖主要放在 Flink lib 中，再通过 HADOOP_CONF_DIR 提供 Hadoop 配置。
手工删除 HDFS 表路径和 checkpoint 目录只是为了保证单机测试可重复执行；生产环境不要直接 hdfs dfs -rm Hudi 表目录或 checkpoint 目录。仅删除 Hudi 表和 checkpoint 不会删除 Kafka 中已有消息，如果重复执行第 5 节 producer，Topic 中会累积重复消息；为了保证结果可复现，测试时建议使用全新的 Kafka Topic，或删除并重建 Topic。
输出中可以看到作业提交成功：
Flink SQL> [INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 83faa7059126b57f0a3c6ab48781a5ef查看 Flink job：
/opt/module/flink-1.20.1/bin/flink list输出：
------------------ Running/Restarting Jobs -------------------
29.09.2025 20:39:40 : 83faa7059126b57f0a3c6ab48781a5ef : insert-into_default_catalog.default_database.hudi_orders_cdc (RUNNING)
--------------------------------------------------------------
No scheduled jobs.因为这是持续流任务，验证完 Hudi commit 后可以取消。取消任务前应先确认 Flink checkpoint 成功，并且 Hudi Timeline 中已经出现 completed commit；否则可能还没有稳定提交到 Hudi 表：
/opt/module/flink-1.20.1/bin/flink cancel 83faa7059126b57f0a3c6ab48781a5ef2.3 验证 Hudi commit 和文件
查看 Timeline：
hdfs dfs -ls /hudi/demo_flink_cdc_orders/.hoodie/timeline完整目录里有 18 个条目，下面只截取两次 commit 相关文件：
Found 18 items
-rw-r--r-- 1 root supergroup 0 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/.hoodie/timeline/20250929203945218.commit.requested
-rw-r--r-- 1 root supergroup 0 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/.hoodie/timeline/20250929203945218.inflight
-rw-r--r-- 1 root supergroup 4344 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/.hoodie/timeline/20250929203945218_20250929203949839.commit
-rw-r--r-- 1 root supergroup 0 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/.hoodie/timeline/20250929203949912.commit.requested
-rw-r--r-- 1 root supergroup 0 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/.hoodie/timeline/20250929203949912.inflight
-rw-r--r-- 1 root supergroup 3933 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/.hoodie/timeline/20250929203949912_20250929203952260.commit在 Hudi 1.0.x 中，完成态 Instant 文件名里可以看到 begin instant 和 completion instant，例如 begin_completion.commit；这里前半段更接近请求 / 开始时间，后半段是完成时间。requested / inflight 用于表达 Instant 状态流转；判断已经成功提交的动作时，主要看完成态 commit 文件。
查看 checkpoint 元信息和数据文件：
hdfs dfs -ls -R /hudi/demo_flink_cdc_orders \
| grep -E 'ckp_meta|2025-09-2[90]|parquet'部分输出：
drwxr-xr-x - root supergroup 0 2025-09-29 20:40 /hudi/demo_flink_cdc_orders/.hoodie/.aux/ckp_meta
-rw-r--r-- 1 root supergroup 0 2025-09-29 20:40 /hudi/demo_flink_cdc_orders/.hoodie/.aux/ckp_meta/20250929204001894.COMPLETED
-rw-r--r-- 1 root supergroup 0 2025-09-29 20:40 /hudi/demo_flink_cdc_orders/.hoodie/.aux/ckp_meta/20250929204001894.INFLIGHT
drwxr-xr-x - root supergroup 0 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/2025-09-28
-rw-r--r-- 1 root supergroup 435862 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/2025-09-28/e7d73c08-2a7f-4be9-aa3f-ac4352592d4d_0-1-0_20250929203945218.parquet
drwxr-xr-x - root supergroup 0 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/2025-09-29
-rw-r--r-- 1 root supergroup 435725 2025-09-29 20:39 /hudi/demo_flink_cdc_orders/2025-09-29/9afc043a-6dc2-4197-8fdd-5170ad042120_0-1-0_20250929203945218.parquet如果要确认第二次 commit 对应的数据文件，可以单独过滤第二个 instant：
hdfs dfs -ls -R /hudi/demo_flink_cdc_orders \
| grep 20250929203949912上面是 grep 截取的部分输出，只展示了第一批 base files；完整排查时应结合第二个 commit 对应的新文件、Timeline 和查询结果一起看。
这里能看到：
- 流式任务按 checkpoint 产生多个 commit。
.hoodie/.aux/ckp_meta中是 Flink Hudi sink 写入过程中使用的 checkpoint / instant 辅助元信息。- 数据按
dt分区写入。 - Hudi 表已经形成可查询的 COW 快照。
.hoodie/.aux/ckp_meta 和 Flink 的 checkpoint 存储目录、Kafka consumer group offset、HoodieStreamer checkpoint 不是同一个概念，不要混淆。数据文件名前半段是 fileId，中间的 0-1-0 类似写入任务相关 token，最后的 20250929203945218 是写入 instant，不要把整个文件名都理解成 fileId。
2.4 查询 CDC 入湖结果
准备查询 SQL。查询脚本使用 batch runtime，只读取已经完成的 Hudi commit，不需要配置 checkpoint。这里不是重新写表，而是在 Flink SQL 中用同样 schema 注册已有 Hudi 路径用于查询；字段、PRIMARY KEY NOT ENFORCED、PARTITIONED BY、table.type 应和已有表配置保持一致。
cat > /tmp/hudi_flink_cdc_query.sql <<'SQL'
SET 'execution.runtime-mode' = 'batch';
SET 'sql-client.execution.result-mode' = 'TABLEAU';
CREATE TABLE hudi_orders_cdc_query (
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_cdc_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 hudi_orders_cdc_query
GROUP BY dt
ORDER BY dt;
SELECT order_id, user_id, amount, status, dt
FROM hudi_orders_cdc_query
ORDER BY order_id;
SQL执行：
/opt/module/flink-1.20.1/bin/sql-client.sh \
-f /tmp/hudi_flink_cdc_query.sql分区统计：
+------------+-----+--------------+--------------+
| dt | cnt | min_order_id | max_order_id |
+------------+-----+--------------+--------------+
| 2025-09-28 | 2 | 12001 | 12002 |
| 2025-09-29 | 1 | 12003 | 12003 |
+------------+-----+--------------+--------------+
2 rows in set (6.85 seconds)明细结果：
+----------+---------+--------+---------+------------+
| order_id | user_id | amount | status | dt |
+----------+---------+--------+---------+------------+
| 12001 | u201 | 39.90 | CREATED | 2025-09-28 |
| 12002 | u202 | 88.00 | PAID | 2025-09-28 |
| 12003 | u203 | 168.00 | CREATED | 2025-09-29 |
+----------+---------+--------+---------+------------+
3 rows in set (1.17 seconds)order_id=12002 从 CREATED 更新成了 PAID，说明 Kafka Debezium JSON 的 update 事件被 Flink SQL 解析成 changelog 后，最终通过 Hudi record key 和 precombine.field=ts 合并成了最新状态。
3. 生产链路能力
3.1 限流：write.rate.limit
CDC 入湖最怕全量快照和实时增量混在一起时出现写入尖峰。比如 Debezium 先扫历史数据，再接 binlog；或者某个任务恢复后大量回放 Kafka offset。这个时候写入端可能瞬间把 HDFS、checkpoint 和 Hudi writer 都打满。
这张图主要说明 write.rate.limit 用来削平 Kafka 堆积追赶和全量快照阶段的写入尖峰，避免 checkpoint、writer buffer 和 HDFS 文件写入同时被打满。
Hudi Flink 提供了 write.rate.limit。本文按 Hudi 1.0.2 Flink connector 语义理解为每秒写入记录数限制；具体单位、默认值和是否对所有 source / sink 场景生效，应以当前版本文档和实测为准：
WITH (
'write.rate.limit' = '1000'
)我对这个参数的理解是：它不是为了提升峰值吞吐，而是为了让写入更平滑。尤其在全量导入、补偿回放、Kafka 堆积追赶这几类场景里，限流可以让 checkpoint 更稳定。
如果没有限流，且 checkpoint 间隔较短、分区较多、write.tasks 较高或 small file 策略不合理，可能出现：
- checkpoint 一直超时。
- writer buffer 快速膨胀。
- HDFS 小文件突然增加。
- Kafka 消费吞吐忽高忽低。
- TaskManager GC 变多。
3.2 离线批量导入
CDC 入湖之前通常需要先做全量导入。全量导入不要直接用和实时任务一样的 upsert 链路硬灌，初始化阶段更适合 bulk_insert。
示例：
SET 'execution.runtime-mode' = 'batch';
SET 'execution.checkpointing.interval' = '0';
CREATE TABLE hudi_orders_full (
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_full',
'table.type' = 'COPY_ON_WRITE',
'write.operation' = 'bulk_insert',
'precombine.field' = 'ts',
'write.tasks' = '4',
'write.bulk_insert.shuffle_input' = 'true',
'write.bulk_insert.sort_input' = 'true',
'write.bulk_insert.sort_input.by_record_key' = 'false'
);本文 Flink 1.20.1 环境中 execution.checkpointing.interval=0 用于关闭 checkpoint；如果你的版本不接受 0，可以直接删除 checkpoint interval 设置，或者按 Flink 文档关闭 checkpoint。这个示例只展示 bulk_insert 操作形态，文件大小调优沿用上一篇参数调优篇，后续应结合 hoodie.parquet.max.file.size、hoodie.parquet.small.file.limit、write.tasks、write.bulk_insert.sort_input、write.bulk_insert.shuffle_input 控制初始化文件布局。
全量导入的核心目标是：
- 快速写入历史数据。
- 尽量生成合理大小的 Parquet 文件。
- 避免一开始就制造大量小文件。
- 为后续实时 upsert 留下清晰的 FileGroup 布局。
这里有两个细节值得单独记：
bulk_insert更适合在 batch runtime mode 下执行，checkpoint 可以关闭，避免初始化任务被流式 checkpoint 开销拖慢。write.tasks会直接影响初始化阶段的输出文件和 FileGroup 数量。当单个文件达到目标大小后还会 roll over，所以最终文件数通常大于等于write.tasks，但不要把这里的文件 / FileGroup 数量和 Bucket Index 的 bucket 概念混在一起。
如果一上来就用小批量 upsert 导历史数据，后面很容易出现写入慢和小文件问题。
3.3 全量接增量与 index bootstrap
全量接增量的核心问题是：实时任务启动时，Hudi 表里已经有历史数据，Flink state 里还没有这些 key 到 FileGroup 的映射。此时如果直接写入增量，可能无法正确识别历史记录位置。
在 Flink Hudi upsert 场景中，index.bootstrap.enabled 用于把已有 Hudi 表中的 record key 位置信息加载进 Flink state。它是 Flink state index 场景下用于已有 Hudi 表接续增量写入的机制，不要和 Spark 侧索引初始化能力混为一谈：
这张图主要说明全量 bulk_insert 完成后，实时 upsert 任务需要通过 index bootstrap 或已有 state 识别历史 record key 的文件位置。
示例配置：
CREATE TABLE hudi_orders_incremental (
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_full',
'table.type' = 'COPY_ON_WRITE',
'write.operation' = 'upsert',
'precombine.field' = 'ts',
'index.bootstrap.enabled' = 'true',
'index.partition.regex' = '2025-09-.*',
'write.index_bootstrap.tasks' = '4'
);这里的 2025-09-.* 是示例分区范围，实际生产要按全量表历史分区范围设置，避免漏加载需要接续的历史分区。
使用 index bootstrap 时要注意：
- 它会把已有表的 key 映射加载到 Flink state。
- 索引加载是阻塞过程，加载期间 checkpoint 可能无法立即成功，需要给失败容忍留余量。
- 可以用
index.partition.regex限定加载分区，避免一次扫全表。 - 首次 bootstrap 完成并确认 checkpoint / savepoint 已经成功保存后，后续从该状态恢复时通常可以关闭它；如果丢失 state 或重新从空状态启动，就需要重新评估是否再次 bootstrap。
官方文档也特别强调：index bootstrap 是阻塞过程，通常要等第一个 checkpoint 成功后，才能认为历史索引加载完成。
实际操作时我会把流程拆得更明确一些：
SET 'execution.checkpointing.interval' = '10s';
SET 'execution.checkpointing.tolerable-failed-checkpoints' = '20';然后按下面的顺序做：
- 第一次启动任务时开启
index.bootstrap.enabled=true。 - 等待日志里出现类似
finish loading the index under partition、Load records from file的加载进度。 - 等第一次 checkpoint 成功，说明历史索引已经进入 Flink state。
- 停止任务并保留 savepoint，或者依赖 externalized checkpoint。
- 第二次启动时关闭
index.bootstrap.enabled，把write.tasks、write.bucket_assign.tasks、write.index_bootstrap.tasks调回生产值。
还有一个容易忽略的点：某些场景下，索引加载进度可能和数据流、分区访问范围、index.partition.regex 以及 checkpoint 推进有关。全量接增量时最好设计一个明确的启动窗口，通过日志和 checkpoint 确认需要接续的分区都已经被覆盖。
4. 常见问题与链路判断
4.1 常见问题
4.1.1 Kafka CDC 表为什么要用 debezium-json
Debezium JSON 中有 before、after、op 等字段，Flink 可以把它解析成 changelog。如果使用普通 kafka connector + json，确实会丢失 Debezium 的 changelog 语义，Hudi 只能看到普通 append 数据；如果使用 upsert-kafka，则是另一种按 key 表达更新的方式，但它和 Debezium before / after / op 原始 JSON 不是同一套格式。
4.1.2 为什么 Hudi 表产生了很多 commit
Flink Hudi sink 按 checkpoint 推进 commit。本文 checkpoint 间隔是 5 秒，且 checkpoint 能持续成功，所以任务运行几十秒后会出现多个 Hudi commit。这是正常现象。生产中要结合延迟、小文件数量和 checkpoint 成功率设置 checkpoint 间隔。
4.1.3 流读任务落后会有什么风险
如果流读任务从很早的 commit 开始消费，但 Hudi clean 已经清掉旧 FileSlice，就可能读不到历史增量。解决方向是：
'read.start-commit' = '20250929203945218'并在写入表配置中合理保留 commit：
clean.retain_commits=30clean.retain_commits 是 Hudi Flink connector 的短名配置，表示保留多少个 commit。清理策略还需要结合 archive、savepoint、下游最大延迟和增量消费 checkpoint 设计。
4.1.4 CDC 入湖能不能直接从 MySQL 读
可以。Flink CDC connector 能直接读取 MySQL binlog。但生产里我更倾向于 MySQL -> Kafka -> Flink -> Hudi，因为 Kafka 提供了回放、削峰、解耦和排查能力。
4.1.5 全量接增量为什么会重复
常见原因是实时写入任务不知道历史 key 已经存在，或者 partition path 发生变化但没有使用合适的索引策略。全量接增量时一定要考虑 index bootstrap、Hudi record key / 业务主键字段、分区字段和 checkpoint 起点。
4.2 CDC 入湖链路判断
CDC 入湖最重要的不是“写进去”，而是保证顺序、去重、回放和恢复。
Kafka 里的 CDC 消息只是原始变化日志；Flink 负责把它变成有状态的 changelog；Hudi 负责把 changelog 落成可查询、可增量消费的湖表。任何一层处理不好，最后都会表现成重复数据、旧值覆盖新值、checkpoint 卡住、下游读不到增量。
我更推荐的入湖顺序是：
- 离线
bulk_insert完成历史数据初始化。 - 实时任务开启
index.bootstrap.enabled识别已有数据。 - 任务第一个 checkpoint 成功后，最好通过 savepoint 或 externalized checkpoint 保留可恢复状态；生产切换配置时优先使用 savepoint 做受控恢复。
- 后续实时 upsert 从 Flink checkpoint / savepoint 中保存的 Kafka offset 和 Hudi writer / index state 继续推进。
- 下游用 Hudi Streaming Read 或 Incremental Query 消费新 commit。
不要把 Kafka consumer group offset、Flink checkpoint 和 Hudi Timeline commit 混为一谈。
这条链路比直接把 binlog 写 Parquet 复杂，但它换来的是可更新、可回放、可增量消费的数据湖表。
5. 总结
本文主要完成了 Flink Hudi 流读、CDC 入湖和全量接增量的梳理：
- Flink Hudi Streaming Read 通过
read.streaming.enabled=true持续读取新 commit。 read.start-commit='earliest'可以从最早可用 commit 开始消费；如果 cleaner 已经清理了旧 FileSlice，落后太久的流读任务仍可能无法完整补读。- Kafka Debezium JSON 可以通过 Flink SQL 写入 Hudi，并保留 update 语义。
write.rate.limit适合全量快照、补偿回放和 Kafka 堆积追赶场景。- 全量导入优先考虑
bulk_insert，实时增量使用upsert。 - 全量接增量的关键是
index.bootstrap.enabled、Hudi record key / partition path 设计，以及 Flink checkpoint / savepoint 状态衔接。
本文以 Hudi 1.0.2 与 Flink 1.20.1 实测环境为准，latest 文档可能已经包含后续版本变化。
