环境准备
Flink 使用 Hudi 前需将 Flink Bundle（本集群使用 hudi-flink1.20-bundle-1.2.0-1.0.0.jar）放入 $FLINK_HOME/lib 目录并重启 Flink 集群，随后在 SQL Client 或 Table API 中以 connector 创建 Hudi 表。
说明
Flink 写 Hudi 的提交由 Checkpoint 驱动：每完成一次 Checkpoint 触发一次 Instant 提交，流式写入场景必须开启 Checkpoint，否则数据不落盘提交。
建表与写入
Flink SQL 建表示例（写入 MOR 表）：
CREATE TABLE orders (
order_id STRING PRIMARY KEY NOT ENFORCED,
order_date STRING,
amount DOUBLE,
ts BIGINT
) PARTITIONED BY (order_date) WITH (
'connector' = 'hudi',
'path' = '<路径>/hudi_db/orders',
'table.type' = 'MERGE_ON_READ',
'write.precombine.field' = 'ts'
);主要WITH 参数说明如下表所示：
| 参数 | 默认值 | 说明 |
|---|---|---|
| connector | 无（必填） | 固定取值hudi |
| path | 无（必填） | 表存储路径（HDFS） |
| table.type | COPY_ON_WRITE | 表类型：COPY_ON_WRITE / MERGE_ON_READ |
| write.operation | upsert | 写入操作：upsert / insert / bulk_insert / insert_overwrite / delete 等 |
| write.precombine.field | ts | 预合并字段（挂靠hoodie.datasource.write.precombine.field） |
| write.tasks | 作业并行度 | 写入算子并行度，默认取执行环境并行度 |
| compaction.tasks | 作业并行度 | 压缩算子并行度（MOR 表） |
| compaction.async.enabled | true | 是否开启异步压缩（MOR 表） |
| hive_sync.enabled | false | 是否同步Hive 元数据（由平台建表流程控制） |
批写入示例（ts 为 BIGINT 时间戳，直接写入整数值即可）：
INSERT INTO orders VALUES ('1001', '2026-09-17', 99.0, 1760000000);更新与删除：upsert 由主键自动完成，删除通过 DELETE FROM 语句实现：
DELETE FROM orders WHERE order_id = '1001';Flink CDC 入湖
常用业务场景：通过Flink CDC 将 MySQL 表数据同步到 Hudi 表，整条链路为 MySQL → Flink CDC → Hudi 表 → Hive/Spark 查询。
先说明mysqlToHudi.sql 的来源：它不是集群自带或框架自带的文件，而是本文档编写的示例脚本——按 step1~step4 的代码，用户在自己的环境（如提交作业的客户端机器）上手动新建一个文本文件，命名为 mysqlToHudi.sql，将四步代码依次粘贴进去保存而成。文件名可自定义，只需与提交命令中 -f 参数指定的路径一致即可。
该脚本的内容按顺序由以下四部分组成：
第1 部分：执行环境参数（SET 语句）；
第2 部分：目标 Hudi 表建表语句（CREATE TABLE，connector=hudi）；
第3 部分：MySQL CDC 源表建表语句（CREATE TABLE，connector=mysql-cdc），字段需与 MySQL 表一致；
第4 部分：同步语句（insert into ... select ...）。
文件结构示意如下（完整内容见step1~step4）：
-- mysqlToHudi.sql 文件内容结构
SET ...; -- step1: 执行环境参数
CREATE TABLE `database_hudi`.`hudi_table` (...); -- step2: 目标表
CREATE TABLE mysql_source (...); -- step3: MySQL源表
insert into `database_hudi`.`hudi_table`
select * from mysql_source; -- step4: 同步语句以下分步说明每一部分的写法（每步的代码块即为需要粘贴进脚本的内容）。
step1：设置执行环境基本参数：
SET 'execution.target' = 'yarn-per-job';
SET execution.checkpointing.interval = 5s;
set restart-strategy = fixed-delay;
set execution.checkpointing.tolerable-failed-checkpoints = 3;
set parallelism.default = 1;
set taskmanager.numberOfTaskSlots = 1;
set jobmanager.memory.process.size = 1g;
set taskmanager.memory.process.size = 2g;
set state.backend = rocksdb;
set state.backend.incremental = true;
set sql-client.execution.result-mode = tableau;注意
Flink 每完成一次 Checkpoint 触发一次 Hudi 提交，Checkpoint 间隔即为数据可见延迟。本示例的 5s 仅用于功能验证，生产环境不建议设置过小：秒级间隔会在 Hudi 侧产生大量小文件，拖慢写入并加剧查询读放大，建议设置为 1~5 分钟。
step2：创建目标 Hudi 表：
create database database_hudi;
CREATE TABLE `database_hudi`.`hudi_table` (
ts BIGINT,
id VARCHAR(40) PRIMARY KEY NOT ENFORCED,
rider VARCHAR(20),
driver VARCHAR(20),
fare DOUBLE,
city VARCHAR(20)
) PARTITIONED BY (`city`) WITH (
'connector' = 'hudi',
'path' = '<路径>/database_hudi/hudi_table',
'table.type' = 'COPY_ON_WRITE',
'cdc.enabled' = 'true'
);step3：创建 MySQL CDC 源表：
CREATE TABLE mysql_source (
ts BIGINT,
id VARCHAR(40),
rider VARCHAR(20),
driver VARCHAR(20),
fare DOUBLE,
city VARCHAR(20),
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = '<mysql-host>',
'port' = '<mysql-port>',
'username' = '<user>',
'password' = '<password>',
'database-name' = '<database>',
'table-name' = '<table>'
);注意
生产环境禁止将数据库密码等敏感信息明文写入SQL 脚本，应通过作业提交参数、配置中心或环境变量注入并对脚本做脱敏管理，避免凭据随脚本分发泄露。
step4：通过 INSERT 方式同步数据：
insert into `database_hudi`.`hudi_table` select * from mysql_source;
step5/step6：启动 Flink 集群并提交作业：
${FLINK_HOME}/bin/start-cluster.sh
${FLINK_HOME}/bin/sql-client.sh -f ${FLINK_HOME}/mysqlToHudi.sql除SQL 方式外，也可使用 Flink CDC 提供的 YAML 作业描述方式（mysql-hudi.yaml，分为 source、sink、pipeline 三段）：
source:
type: mysql
hostname: <mysql-host>
port: <mysql-port>
username: <user>
password: <password>
tables: <database>.<table>
sink:
type: hudi
tablename: hudi_table
pk: id
partition.field: city
tablepath: <路径>/database_hudi/hudi_table
pipeline:
name: Sync MySQL Database to Hudi
parallelism: 4
flink-conf.execution.checkpointing.timeout: 47min
flink-conf.execution.checkpointing.interval: 1 min其中source 部分配置订阅的 MySQL 连接信息与表信息；sink 部分描述 Hudi 目标表（表名、路径、主键 pk、分区字段）；pipeline 部分描述作业名称、并行度与 checkpoint 配置。执行作业：
${FLINKCDC_HOME}/bin/flink-cdc.sh ./mysql-hudi.yaml --yarn-application入湖完成后，可通过Spark 引擎或 Hive 引擎进行查询。
查询数据
Flink SQL 读取 Hudi 表支持批与流两种模式，MOR 表流读通过读取日志尾部增量记录实现近实时消费：
-- 批（快照）读取
SELECT * FROM orders WHERE order_date = '2026-09-17';
-- 流式读取（读取表增量，持续输出新写入数据）
SET execution.runtime-mode = streaming;
SELECT * FROM orders /*+ OPTIONS('read.streaming.enabled'='true',
'read.start-commit'='20260917093000123') */;流读相关主要参数：read.streaming.enabled（是否流读）、read.start-commit（起始消费时刻）、read.streaming.check-interval（增量检查间隔，默认 30 秒）。
Flink 集成注意事项
1、流式写入必须开启 Checkpoint，Checkpoint 间隔即为数据可见延迟；
2、MOR 表建议保持默认异步压缩（compaction.async.enabled=true），避免日志堆积拖慢快照查询；
3、同一张表不可同时被 Spark 与 Flink 写入，跨引擎写入会因并发提交冲突失败；读写可跨引擎（Flink 写、Spark 读或反之）；
4、Flink 写入无推测执行问题（推测执行为 Spark 特有），Spark 侧写入的推测执行限制见"使用说明-环境准备"。
