Apache Hudi 学习笔记 10：Flink SQL 集成 Hudi 实战

hudi

发布日期: 2025-09-07

文章字数: 6.8k

阅读时长: 30 分

阅读次数:

前面几篇更多是从 Spark 的角度使用 Hudi。Spark 很适合离线初始化、批量修复、交互式 SQL 和批处理任务，但如果要把 MySQL CDC、Kafka 事件流这类实时数据持续写入 Hudi，我更倾向于用 Flink。

Flink 写 Hudi 的关键不是“能不能执行一条 INSERT”，而是 checkpoint、state backend、bucket assigner、writer task、Timeline commit 这些组件能不能配合起来。本文用 Flink SQL 跑一条真实写入链路：创建 Hudi COW 表、写入订单数据、通过 checkpoint 形成 Hudi commit，再用 Flink SQL 查询结果。

1. Flink SQL 接入环境

1.1 背景问题

假设我们有一张订单表，MySQL binlog 或 Kafka 里不断产生下面几类事件：

新订单创建。
订单状态更新，比如 CREATED 变成 PAID。
订单按天分区落入数据湖。
查询侧希望看到最新状态。

如果只是把 Kafka 数据写成普通 Parquet 文件，更新语义很难处理。同一个 order_id 可能出现多次，查询时还要自己按时间取最新。Hudi 的价值在于把这种 upsert 语义下沉到表管理层，而 Flink 的价值在于持续消费、状态管理和 checkpoint 提交。

Flink SQL 写 Hudi 的主流程可以理解成下面这样：

Flink SQL 写入 Hudi 的主流程

这张图主要说明 Flink SQL 写入 Hudi 时，SQL DDL / DML 只是入口，真正的提交需要通过 Flink checkpoint 协调 Hudi Timeline。

这里最重要的一点是：Flink 写 Hudi 不是写完文件就结束。Hudi commit 需要和 Flink checkpoint 协调，checkpoint 成功后，Hudi Timeline 中对应的 Instant 才进入 completed 状态，查询侧才稳定可见。

1.2 版本和目录检查

先确认当前 Flink 和 Hudi bundle：

/opt/module/flink-1.20.1/bin/flink --version

ls -lh \
  /opt/module/hudi-1.0.2/lib/hudi-flink1.20-bundle-1.0.2.jar

本文环境中实际使用的 Hudi Flink bundle：

-rw-r--r-- 1 root root 57M Jun 15 21:37 /opt/module/hudi-1.0.2/lib/hudi-flink1.20-bundle-1.0.2.jar

Flink SQL Client 路径是：

/opt/module/flink-1.20.1/bin/sql-client.sh

注意不要写成 /opt/module/flink/bin/sql-client.sh。我这台机器实际安装目录就是 /opt/module/flink-1.20.1，文章里的命令都按这个路径来。

1.3 配置 Flink Hudi bundle

Hudi Flink SQL connector 需要让 Flink SQL Client 和 Flink runtime 都能加载到。我的处理方式是把 bundle 放到 Flink lib 目录：

cp -f \
  /opt/module/hudi-1.0.2/lib/hudi-flink1.20-bundle-1.0.2.jar \
  /opt/module/flink-1.20.1/lib/

如果改了 Flink lib，必须重启 Flink 集群：

/opt/module/flink-1.20.1/bin/stop-cluster.sh
/opt/module/flink-1.20.1/bin/start-cluster.sh

验证 Flink Web UI：

Flink Web UI

这张截图主要用于确认 Flink 1.20.1 集群已经启动，并且后续 Hudi 写入 / 查询任务能够在 Web UI 中看到执行记录。这个截图里能看到 Flink 版本是 1.20.1，Completed Job 中有后面写入和查询 Hudi 表的记录。

1.4 Hadoop 依赖和 classloader 处理

这一步是本文最容易踩坑的地方。

Flink 发行包默认不一定包含 Hadoop 依赖。如果只设置 HADOOP_CONF_DIR，但 Flink lib 中没有 Hadoop jar，执行 Hudi HDFS 写入时会报：

java.lang.ClassNotFoundException: org.apache.hadoop.conf.Configuration

本文 Hadoop 安装目录按 /opt/module/hadoop-3.4.1 记录；如果你的机器使用 /opt/hadoop 或 /opt/hadoop-3.4.1 软链接，按实际路径替换即可。我补充了 Hadoop 3.4.1 的运行依赖到 Flink lib：

cp -f /opt/module/hadoop-3.4.1/share/hadoop/common/hadoop-common-3.4.1.jar \
  /opt/module/flink-1.20.1/lib/

cp -f /opt/module/hadoop-3.4.1/share/hadoop/common/lib/hadoop-auth-3.4.1.jar \
  /opt/module/flink-1.20.1/lib/

cp -f /opt/module/hadoop-3.4.1/share/hadoop/hdfs/hadoop-hdfs-client-3.4.1.jar \
  /opt/module/flink-1.20.1/lib/

cp -f /opt/module/hadoop-3.4.1/share/hadoop/mapreduce/hadoop-mapreduce-client-core-3.4.1.jar \
  /opt/module/flink-1.20.1/lib/

cp -f /opt/module/hadoop-3.4.1/share/hadoop/yarn/hadoop-yarn-api-3.4.1.jar \
  /opt/module/flink-1.20.1/lib/

同时复制 Hadoop common/lib 中的常用依赖，但排除了 SLF4J、reload4j、metrics 这类容易和 Flink 冲突的 jar：

for jar in /opt/module/hadoop-3.4.1/share/hadoop/common/lib/*.jar; do
  name=$(basename "$jar")
  case "$name" in
    slf4j-*|metrics-core-*|reload4j-*|log4j-*|hadoop-auth-3.4.1.jar)
      ;;
    *)
      cp -f "$jar" /opt/module/flink-1.20.1/lib/
      ;;
  esac
done

这一步是本文单机环境的兼容处理。生产环境更推荐使用与 Hadoop 版本匹配的 Flink Hadoop 发行包、配置 yarn.provided.lib.dirs，或者通过镜像 / 依赖管理统一 Flink runtime 依赖，不建议随意把 Hadoop lib 全量复制到 Flink lib。如果复制后出现新的 LinkageError、NoSuchMethodError、ClassCastException，需要继续排查 Flink lib 中是否出现 Guava、Jackson、Netty、Curator、metrics 等重复版本。本文命令只针对当前机器实测通过，不是通用依赖白名单。

这台机器还遇到了 com.codahale.metrics.Meter 类加载冲突：

java.lang.LinkageError: loader constraint violation:
loader 'app' wants to load class com.codahale.metrics.Meter.
A different class with the same name was previously loaded by
org.apache.flink.util.ChildFirstClassLoader

原因是 Flink 1.20.1 的 flink-dist 和 Hudi Flink bundle 中都带了同名 metrics 类。下面是本文单机环境的临时修复，不建议作为生产标准做法。更稳妥的优先级是：先确认 Flink lib 中是否混入重复 metrics jar；再评估 classloader parent-first / child-first；优先使用官方匹配 bundle 或重新 shade；只有单机测试无法继续时，才临时删除 bundle 内冲突类，并保留 .orig 备份。

cd /opt/module/flink-1.20.1/lib

cp -f \
  /opt/module/hudi-1.0.2/lib/hudi-flink1.20-bundle-1.0.2.jar \
  hudi-flink1.20-bundle-1.0.2.jar.orig

cp -f hudi-flink1.20-bundle-1.0.2.jar.orig \
  hudi-flink1.20-bundle-1.0.2.jar

zip -q -d hudi-flink1.20-bundle-1.0.2.jar 'com/codahale/metrics/*'

jar tf hudi-flink1.20-bundle-1.0.2.jar \
  | grep 'com/codahale/metrics/Meter.class' || echo 'metrics classes removed'

输出：

metrics classes removed

然后在 Flink 1.20.1 的 config.yaml 中增加 parent-first 配置：

classloader:
  resolve:
    order: parent-first

这个配置会影响整个 Flink 集群的类加载顺序，不是只影响 Hudi connector。生产环境应在独立测试集群验证，避免影响其他 Flink 作业。Flink 1.20 的默认配置文件可以是 config.yaml；如果你的发行包仍使用 flink-conf.yaml 的 key-value 写法，应改成对应格式，例如 classloader.resolve-order: parent-first 或当前版本文档要求的键名。具体以当前 Flink 发行包配置文件为准。

再次重启 Flink：

/opt/module/flink-1.20.1/bin/stop-cluster.sh
/opt/module/flink-1.20.1/bin/start-cluster.sh

这不是一个通用“必须删除 jar 内容”的建议，而是这套单机环境真实遇到的版本冲突处理。生产环境更推荐使用官方匹配的二进制包、统一依赖来源，或者在构建阶段解决 shading 冲突。

2. 写入和查询实践

2.1 Flink 写 Hudi 的运行组件

Flink 写 Hudi 时，SQL DDL 只是入口。真正运行起来以后，会有 source、bucket assigner、stream writer、checkpoint coordinator 等组件参与。

Flink 写 Hudi 的运行组件

这张图主要说明 Flink Hudi Sink 的运行时组件，包括 bucket assigner、writer task、checkpoint coordinator 和 Hudi commit。

几个核心点：

PRIMARY KEY NOT ENFORCED 定义 Hudi record key。
precombine.field 决定同一个 key 多条记录谁更新谁。
write.bucket_assign.tasks 负责把记录分配到 FileGroup。
write.tasks 负责实际写文件。
checkpoint 成功后，Hudi commit 才进入 completed。
单机测试时并行度不能盲目调大，否则 slot 不够会让 job 卡在 CREATED / SCHEDULED。

本文的 VALUES source 是小规模有界输入，作业结束也会推进最终写入提交；长期 Kafka CDC 流任务中更应关注持续 checkpoint 是否稳定成功。

2.2 准备 Flink SQL 写入脚本

本文主线使用 streaming runtime，是为了按实时入湖方式验证 checkpoint 与 Hudi commit 的协调。Flink Hudi 也存在 batch / bounded 写入场景，具体运行模式要按 source 类型和写入需求确认。测试数据用 VALUES source，一次写入 5 条订单，其中 order_id=10002 出现两次，后一次状态是 PAID，时间戳更大。

这里的 dt 是模拟业务分区日期，用于演示多分区写入，不代表文章实际写作日期。ts 是模拟业务事件时间，用作 precombine 排序字段；dt 是模拟业务分区字段，本文只用于验证多分区写入。

cat > /tmp/hudi_flink_sql_stream_insert.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_stream_demo';
SET 'state.backend.type' = 'rocksdb';

DROP TABLE IF EXISTS demo_flink_hudi_orders;

CREATE TABLE demo_flink_hudi_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_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',
  'read.tasks' = '1'
);

INSERT INTO demo_flink_hudi_orders VALUES
  (10001, 'u001', CAST(99.90 AS DECIMAL(10,2)), 'CREATED', 1757149200000, '2025-09-06'),
  (10002, 'u002', CAST(168.00 AS DECIMAL(10,2)), 'CREATED', 1757149260000, '2025-09-06'),
  (10003, 'u003', CAST(39.90 AS DECIMAL(10,2)), 'CREATED', 1757149320000, '2025-09-07'),
  (10002, 'u002', CAST(168.00 AS DECIMAL(10,2)), 'PAID', 1757149500000, '2025-09-06'),
  (10004, 'u004', CAST(258.00 AS DECIMAL(10,2)), 'CREATED', 1757149560000, '2025-09-07');
SQL

这里有两个配置我单独说明一下。

第一，state.backend.type 这种包含 state 的配置在 Flink SQL 文件里最好写成带引号的形式：

SET 'state.backend.type' = 'rocksdb';

如果写成下面这样，SQL Parser 可能直接报语法错误：

SET state.backend.type = rocksdb;

实际错误如下：

org.apache.flink.sql.parser.impl.ParseException:
Encountered "state" at line 1, column 5.

本文 Flink 1.20.1 环境中 SET 'state.backend.type' = 'rocksdb' 实测可用；如果启动时报 state backend 配置错误，应按当前 Flink 1.20 配置文档检查取值是否需要写成 embedded-rocksdb 或其他形式。

这里的 DROP TABLE 主要清理 Flink SQL 当前 Catalog 中的表定义；为了保证测试数据重跑一致，本文还会手工删除 HDFS 表路径。生产环境不要直接 hdfs dfs -rm Hudi 表目录。本文显式写出 datasource 风格 record key / partition path 配置，是为了让 Hudi 表属性更清晰；在 Flink SQL 中部分信息也可以从 PRIMARY KEY 和 PARTITIONED BY 推断，具体以 Hudi 1.0.2 Flink connector 行为为准。本文关闭 metadata table 是为了减少单机测试变量；生产环境是否开启要结合表规模、文件数量、查询方式和 Hudi / Flink connector 版本评估。

第二，本文把 write.tasks 和 write.bucket_assign.tasks 都设置为 1，不是因为生产环境应该这么配，而是单机测试环境 slot 有限。之前设置成 2 时，Flink Web UI 中任务一直卡住，因为需要的 subtask 数超过可用 slot。

2.2.1 Flink SQL 的更新语义

在本文的 Flink Hudi 流式 upsert 场景中，更新更常见的表达方式不是手写 UPDATE SQL，而是再次写入相同主键、更大 precombine.field 的变更记录，让 Hudi 在写入阶段完成合并。Flink SQL DML 的 UPDATE / DELETE 支持情况需要按 Hudi 1.0.2 Flink connector 文档和实测确认。

例如下面这条数据会把 order_id=10001 的状态更新为 CANCELLED：

INSERT INTO demo_flink_hudi_orders VALUES
  (10001, 'u001', CAST(99.90 AS DECIMAL(10,2)), 'CANCELLED', 1757149800000, '2025-09-06');

这和普通数据库的 UPDATE table SET ... WHERE ... 不完全一样。Flink SQL 负责把变更记录送进 Hudi Sink，Hudi 根据 PRIMARY KEY NOT ENFORCED 和 precombine.field 判断同一条业务记录应该保留哪个版本。

所以这里有两个约束：

Hudi record key / 业务主键字段必须稳定，不能一会儿用 order_id，一会儿用组合字段；这里的 PRIMARY KEY NOT ENFORCED 不是数据库强约束。
precombine.field 要能表达事件先后顺序，比如业务更新时间、binlog 时间或递增版本号。

2.2.2 datagen 流式插入示例

如果只是验证 Flink Hudi Sink 是否能持续接收流式数据，可以用 Flink 内置的 datagen connector 造数。下面这个例子不作为本文主线实测结果，只是补充一个持续写入模板：

SET 'execution.runtime-mode' = 'streaming';
SET 'sql-client.execution.result-mode' = 'TABLEAU';
SET 'execution.checkpointing.interval' = '10s';
SET 'execution.checkpointing.dir' = 'hdfs:///flink-checkpoints/hudi_datagen_demo';
SET 'state.backend.type' = 'rocksdb';

CREATE TABLE source_orders (
  order_id BIGINT,
  user_id STRING,
  amount DECIMAL(10,2),
  status STRING,
  ts BIGINT,
  dt AS CAST('2025-09-06' AS STRING)
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1',
  'fields.order_id.kind' = 'sequence',
  'fields.order_id.start' = '20001',
  'fields.order_id.end' = '20100',
  'fields.ts.kind' = 'sequence',
  'fields.ts.start' = '1757149200000',
  'fields.ts.end' = '1757152800000',
  'fields.user_id.length' = '4',
  'fields.status.length' = '6'
);

CREATE TABLE demo_flink_hudi_orders_stream (
  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_orders_stream',
  'table.type' = 'COPY_ON_WRITE',
  'write.operation' = 'upsert',
  'precombine.field' = 'ts',
  'hoodie.metadata.enable' = 'false',
  'write.tasks' = '1',
  'write.bucket_assign.tasks' = '1'
);

INSERT INTO demo_flink_hudi_orders_stream
SELECT order_id, user_id, amount, status, ts, dt
FROM source_orders;

datagen 很适合验证 checkpoint、writer task 和文件增长，但它生成的是随机或规则数据，不能替代真实 CDC 测试。这里的 status 只是随机字符串，不代表真实订单状态；这个例子只用于验证持续写入、checkpoint 和文件增长，不用于验证业务枚举语义。如果 datagen 用来验证 upsert 更新语义，需要为 ts 配置可控生成规则；本文这个 datagen 模板只验证持续写入，不验证同 key 更新顺序。

2.3 执行 Flink SQL 写入

清理旧表和 checkpoint：

hdfs dfs -rm -r -f \
  /hudi/demo_flink_hudi_orders \
  /flink-checkpoints/hudi_sql_stream_demo

本文使用默认 / 临时 Catalog，所以 DROP TABLE + 删除 HDFS 路径能保证单机测试重跑。若使用 Hive Catalog 或持久化 Catalog，删除表定义和删除 HDFS 数据要分别确认。

执行 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_sql_stream_insert.sql

这里 unset HADOOP_CLASSPATH 是为了避免把本机 Hadoop 依赖再次通过环境变量注入 Flink，造成和 Flink lib 中已有依赖重复。本文依赖主要放在 Flink lib 中，再通过 HADOOP_CONF_DIR 提供 Hadoop 配置。

输出中可以看到 job 提交成功：

Flink SQL>
> INSERT INTO demo_flink_hudi_orders VALUES
>   (10001, 'u001', CAST(99.90 AS DECIMAL(10,2)), 'CREATED', 1757149200000, '2025-09-06'),
>   (10002, 'u002', CAST(168.00 AS DECIMAL(10,2)), 'CREATED', 1757149260000, '2025-09-06'),
>   (10003, 'u003', CAST(39.90 AS DECIMAL(10,2)), 'CREATED', 1757149320000, '2025-09-07'),
>   (10002, 'u002', CAST(168.00 AS DECIMAL(10,2)), 'PAID', 1757149500000, '2025-09-06'),
>   (10004, 'u004', CAST(258.00 AS DECIMAL(10,2)), 'CREATED', 1757149560000, '2025-09-07')
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: a7ddebc5c86313940da40cb98504089b

查看 Flink job：

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

写入完成后没有运行中的 job：

No running jobs.
No scheduled jobs.

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

2.4 查看 Hudi Timeline 和文件

查看 Hudi Timeline：

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

输出如下：

Found 4 items
-rw-r--r--   1 root supergroup          0 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/.hoodie/timeline/20250907152932141.commit.requested
-rw-r--r--   1 root supergroup          0 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/.hoodie/timeline/20250907152932141.inflight
-rw-r--r--   1 root supergroup       4372 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/.hoodie/timeline/20250907152932141_20250907152934179.commit
drwxr-xr-x   - root supergroup          0 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/.hoodie/timeline/history

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

查看数据文件：

hdfs dfs -ls -R /hudi/demo_flink_hudi_orders \
  | grep -E '2025-09-0[78]|parquet|ckp_meta'

输出如下：

drwxr-xr-x   - root supergroup          0 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/.hoodie/.aux/ckp_meta
-rw-r--r--   1 root supergroup          0 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/.hoodie/.aux/ckp_meta/20250907152932141.COMPLETED
-rw-r--r--   1 root supergroup          0 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/.hoodie/.aux/ckp_meta/20250907152932141.INFLIGHT
drwxr-xr-x   - root supergroup          0 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/2025-09-06
-rw-r--r--   1 root supergroup     435903 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/2025-09-06/4910892f-4e0e-47f4-8364-da0110e33d9e_0-1-0_20250907152932141.parquet
drwxr-xr-x   - root supergroup          0 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/2025-09-07
-rw-r--r--   1 root supergroup     435972 2025-09-07 15:29 /hudi/demo_flink_hudi_orders/2025-09-07/2ff86876-7b5e-404f-9a30-9514b4471312_0-1-0_20250907152932141.parquet

可以看到：

Timeline 中生成了一个 completed commit。
.hoodie/.aux/ckp_meta 中有 Flink Hudi 写入过程使用的 checkpoint / instant 辅助元信息。它和前一篇 HoodieStreamer 的 Kafka source checkpoint 不是同一个概念，不要混淆。
两个分区分别生成了 Parquet 文件。
文件名里带有同一个 instant：20250907152932141。文件名前半段是 fileId，中间的 0-1-0 类似写入任务相关 token，最后的 20250907152932141 是写入 instant，不要把整个文件名都理解成 fileId。

下面这张图对应的就是 checkpoint 到 Hudi commit 的关系：

Checkpoint 驱动 Hudi Commit

这张图主要说明 checkpoint 成功后，Hudi commit 才会稳定进入 completed 状态，查询侧才能基于 completed instant 构造可见快照。

2.5 使用 Flink SQL 查询 Hudi 表

准备查询脚本：

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

CREATE TABLE demo_flink_hudi_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_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_orders
GROUP BY dt
ORDER BY dt;

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

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

执行：

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_sql_query.sql

分区统计结果：

+------------+-----+--------------+--------------+
|         dt | cnt | min_order_id | max_order_id |
+------------+-----+--------------+--------------+
| 2025-09-06 |   2 |        10001 |        10002 |
| 2025-09-07 |   2 |        10003 |        10004 |
+------------+-----+--------------+--------------+
2 rows in set (6.15 seconds)

明细查询结果：

+----------+---------+--------+---------+------------+
| order_id | user_id | amount |  status |         dt |
+----------+---------+--------+---------+------------+
|    10001 |    u001 |  99.90 | CREATED | 2025-09-06 |
|    10002 |    u002 | 168.00 |    PAID | 2025-09-06 |
|    10003 |    u003 |  39.90 | CREATED | 2025-09-07 |
|    10004 |    u004 | 258.00 | CREATED | 2025-09-07 |
+----------+---------+--------+---------+------------+
4 rows in set (1.18 seconds)

order_id=10002 最终状态是 PAID，说明 Hudi 根据 Flink SQL 中声明的 PRIMARY KEY NOT ENFORCED，也就是 Hudi record key，以及 precombine.field=ts 做了更新合并。

3. 部署与类型补充

3.1 yarn-session 模式说明

本文实际跑的是 standalone 单机模式。生产或更接近集群测试时，可以使用 yarn-session、per-job 或 application mode。本文只简单给出 yarn-session 形态，具体部署模式要按集群标准选择：

/opt/module/flink-1.20.1/bin/yarn-session.sh \
  -jm 1024m \
  -tm 2048m \
  -s 2 \
  -d

-s 2 表示每个 TaskManager 的 slot 数，实际值要结合 write.tasks、write.bucket_assign.tasks、source 并行度和集群资源设置。

然后再启动 SQL Client：

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

/opt/module/flink-1.20.1/bin/sql-client.sh

如果使用 yarn-session，要确认 Hudi bundle、Hadoop 依赖、Flink 配置在提交到 YARN 的环境中也能被 TaskManager 读取到。只在客户端机器上 -j 一个 jar，有时 SQL Client 能识别 connector，但真正的 TaskManager 不一定有同样的依赖。这也是本文第 3 节选择把 Hudi bundle 放入 Flink lib，而不是长期依赖 sql-client -j 的原因。

3.2 Java 程序提交方式

除了 SQL Client，也可以用 Java 程序创建 StreamTableEnvironment 后执行同样的 DDL / DML。核心依赖节选如下，不是完整 POM：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-table-api-java-bridge</artifactId>
  <version>1.20.1</version>
</dependency>

<dependency>
  <groupId>org.apache.hudi</groupId>
  <artifactId>hudi-flink1.20-bundle</artifactId>
  <version>1.0.2</version>
  <scope>provided</scope>
</dependency>

完整 Maven 依赖还需要根据 Flink 1.20.1 的 Table / Streaming API、planner / runtime 和提交方式确认。Hudi bundle 如果已经放在 Flink lib，业务 jar 中通常用 provided。本文以 Hudi 1.0.2 的 hudi-flink1.20-bundle 为准；如果使用 Hudi 1.1+ 或其他 Flink 版本，要回到对应版本 Maven Central / 官方文档核对 artifactId。

程序结构大致如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5000);

EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode()
    .build();

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

tableEnv.executeSql("CREATE TABLE demo_flink_hudi_orders (...) WITH (...)");
tableEnv.executeSql("INSERT INTO demo_flink_hudi_orders SELECT ...");

这只是结构示意。实际程序中需要保存 TableResult，并根据任务类型决定是否调用 await，或者由 flink run 提交后在集群中长期运行。

如果是长期运行的流式任务，不建议只写一行 enableCheckpointing。至少要把 checkpoint 目录、语义、超时时间和外部化 checkpoint 配清楚：

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

Configuration conf = new Configuration();
conf.setString("state.backend.type", "rocksdb");
conf.setString("execution.checkpointing.dir", "hdfs:///flink/checkpoints/hudi_java_demo");

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

CheckpointConfig checkpointConfig = env.getCheckpointConfig();
checkpointConfig.setMinPauseBetweenCheckpoints(3000);
checkpointConfig.setCheckpointTimeout(Duration.ofMinutes(1).toMillis());
checkpointConfig.setTolerableCheckpointFailureNumber(3);
checkpointConfig.enableExternalizedCheckpoints(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
);

EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode()
    .build();

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

这里的 checkpoint 目录必须是 TaskManager 能访问的持久化路径。单机调试可以用本地路径，和 Hudi/HDFS 联调时更建议直接放到 HDFS。

Java API 以 Flink 1.20.1 为准；如果编译时报 API 变化，按当前 Flink 版本的 CheckpointConfig 文档调整。

提交命令：

/opt/module/flink-1.20.1/bin/flink run \
  -c cn.hnbian.hudi.FlinkHudiSqlDemo \
  /opt/app/hudi-flink-demo.jar

如果是 YARN per-job，可以这样提交：

/opt/module/flink-1.20.1/bin/flink run \
  -t yarn-per-job \
  -c cn.hnbian.hudi.FlinkHudiSqlDemo \
/opt/app/hudi-flink-demo.jar

不同 Flink 版本和集群发行版对 yarn-per-job / yarn-application / session 的推荐不同，本文只是给出命令形态；生产提交模式以当前集群标准为准。

如果 Hudi bundle 已经放在 Flink lib，业务 jar 中不要再打入一份完整 Hudi bundle，否则很容易再次遇到类冲突。

3.3 Flink SQL 类型映射

Flink SQL 写入 Hudi 时，最终底层仍然要落到 Hudi 表的 Avro schema 和 Parquet/Log 文件里。类型映射如果不清楚，最容易在 DECIMAL、TIMESTAMP、DATE 这些字段上踩坑。

常用类型可以按下面理解：

Flink SQL Type	Hudi / Avro 表达	使用建议
`CHAR` / `VARCHAR` / `STRING`	`string`	业务 ID、枚举、状态字段最常用
`BOOLEAN`	`boolean`	布尔标记
`TINYINT` / `SMALLINT` / `INT`	`int`	普通整数
`BIGINT`	`long`	主键、毫秒时间戳、计数字段
`FLOAT`	`float`	一般业务较少用
`DOUBLE`	`double`	浮点指标
`DECIMAL(p,s)`	Avro decimal logical type	金额字段建议用 `DECIMAL`，不要用 `DOUBLE`
`DATE`	Avro date logical type	日期字段，常用于分区或统计日期
`TIME`	Avro time logical type	单独使用较少
`TIMESTAMP(3)`	Avro timestamp logical type	事件时间、更新时间
`ARRAY<T>`	Avro array	嵌套字段需要确认查询引擎兼容性
`MAP<K,V>`	Avro map	key 通常为 string 更稳
`ROW<...>`	Avro record	复杂结构建议先做小表验证

这篇主线里我把 ts 写成 BIGINT，是为了让 precombine.field 的比较非常直接。也可以写成 TIMESTAMP(3)，但要确认上游时间语义和 Flink/Hudi 类型转换一致。如果用 TIMESTAMP(3) 做 precombine.field，需要单独验证同 key 多版本比较结果，尤其是时区、精度和 null 值处理。

金额字段我使用：

amount DECIMAL(10,2)

而不是：

amount DOUBLE

原因是订单金额属于精确数值，DOUBLE 更适合指标近似计算，不适合保存交易金额。

分区字段这里使用字符串日期：

dt STRING

这是为了让 HDFS 分区目录直观稳定，比如 /hudi/demo_flink_hudi_orders/2025-09-06。如果 DATE 类型直接作为分区字段，需要额外确认 Hudi 分区路径格式、Hive Sync 类型映射和下游引擎识别方式；本文主线用 STRING 是为了路径稳定直观。

4. 常见问题与接入判断

4.1 常见问题

4.1.1 SQL Client 能建表，但 INSERT 报 ClassNotFoundException

现象：

java.lang.ClassNotFoundException: org.apache.hadoop.conf.Configuration

原因通常是 Flink runtime 缺 Hadoop 依赖。解决方式是补 Hadoop jar 到 Flink lib，或者使用和集群匹配的 Flink Hadoop 发行包，并设置：

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

4.1.2 bundle 通过 `-j` 加载后作业端找不到 jar

现象：

java.nio.file.NoSuchFileException:
/opt/module/flink-1.20.1/lib/hudi-flink1.20-bundle-1.0.2.jar

如果使用 sql-client.sh -j 指定本地 jar，SQL Client 端可能能识别 connector，但 TaskManager 端未必能访问同一路径，可能出现 NoSuchFileException 或 ClassNotFoundException。本文最终采用的是把 Hudi Flink bundle 放进 Flink lib，并重启集群。

4.1.3 不要重复加载 Hudi bundle

如果既放进 Flink lib，又在 sql-client.sh -j 里再传一次同一个 bundle，可能出现重复类加载问题。本文最终使用的是：

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

也就是不再额外加 -j。

4.1.4 非交互模式查询必须使用 TABLEAU

如果在 sql-client.sh -f 中执行查询，结果模式需要设置为 TABLEAU：

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

否则可能报：

In non-interactive mode, it only supports to use TABLEAU as value of
sql-client.execution.result-mode when execute query.

4.1.5 slot 不够会导致 job 一直不动

本文一开始把 write.tasks 和 write.bucket_assign.tasks 都设为 2，单机可用 slot 不够，Flink Web UI 里 job 一直处于 RUNNING，但顶点是 CREATED / SCHEDULED，没有实际写数据。

可以在 Flink Web UI 的 Job Graph / SubTasks 页面查看每个 vertex 的状态，确认是否因为 slot 不足没有被调度执行。

单机测试可以先降到：

'write.tasks' = '1',
'write.bucket_assign.tasks' = '1',
'read.tasks' = '1'

生产环境再结合 TaskManager 数量、slot 数、Kafka 分区数、写入吞吐逐步调大。

4.2 Flink SQL 接入判断

Flink 集成 Hudi 的重点不是 SQL 语法，而是运行时。Spark 写 Hudi 更像一次批处理提交，Flink 写 Hudi 更像一个持续运行的事务协调系统：数据先进入 state 和 writer，checkpoint 成功后再推进 Hudi Timeline。本文 VALUES 小样本是为了快速验证链路；真正的 Kafka CDC 长期流任务中，checkpoint 是否持续成功才是 Hudi commit 稳定推进的关键。

这也是为什么 Flink Hudi 的问题经常出在这些地方：

bundle 没放对位置。
TaskManager classpath 和 SQL Client classpath 不一致。
checkpoint 没开或 checkpoint 不成功。
并行度超过 slot，任务看起来 RUNNING，实际没有写。
precombine.field 选错，导致同一个 key 的更新顺序不符合预期。

我更建议先用很小的数据、很低的并行度把“写入、commit、查询”跑通，再逐步增加 Kafka source、CDC 格式、并行度、compaction 和 Hive Sync。否则一上来就把 CDC、Kafka、Hudi、Hive、YARN、checkpoint 全放进去，排查会非常混乱。

5. 总结

本文用 Flink SQL 跑通了 Hudi COW 表写入和查询：

Hudi Flink bundle 使用 hudi-flink1.20-bundle-1.0.2.jar。
bundle 放到 Flink lib 后必须重启 Flink。
Flink 流式写入 Hudi 要重点关注 checkpoint、state backend 和 writer 并行度；batch / bounded 写入还需要结合 source 类型和作业结束提交语义单独验证。
单机测试不要盲目调大 write.tasks，slot 不够会让任务卡住。
Hudi commit 形成后，可以在 .hoodie/timeline 中看到 completed commit。
Flink SQL 查询结果验证了 order_id=10002 的 upsert 更新生效。
DECIMAL、TIMESTAMP、DATE 等类型要结合 Avro logical type 和下游查询引擎一起验证。

hnbian

https://www.hnbian.cn/posts/f4d8a2b7.html

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !

flink hudi data-lake flink-sql checkpoint rocksdb

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

2025-09-18 hudi

flink hudi compaction checkpoint tuning performance

Apache Hudi 学习笔记 09：并发控制、锁机制与 Spark 写入调优

2025-08-28 hudi

spark hudi concurrency tuning lock performance

Apache Hudi 学习笔记 10：Flink SQL 集成 Hudi 实战

1. Flink SQL 接入环境

1.1 背景问题

1.2 版本和目录检查

1.3 配置 Flink Hudi bundle

1.4 Hadoop 依赖和 classloader 处理

2. 写入和查询实践

2.1 Flink 写 Hudi 的运行组件

2.2 准备 Flink SQL 写入脚本

2.2.1 Flink SQL 的更新语义

2.2.2 datagen 流式插入示例

2.3 执行 Flink SQL 写入

2.4 查看 Hudi Timeline 和文件

2.5 使用 Flink SQL 查询 Hudi 表

3. 部署与类型补充

3.1 yarn-session 模式说明

3.2 Java 程序提交方式

3.3 Flink SQL 类型映射

4. 常见问题与接入判断

4.1 常见问题

4.1.1 SQL Client 能建表，但 INSERT 报 ClassNotFoundException

4.1.2 bundle 通过 -j 加载后作业端找不到 jar

4.1.3 不要重复加载 Hudi bundle

4.1.4 非交互模式查询必须使用 TABLEAU

4.1.5 slot 不够会导致 job 一直不动

4.2 Flink SQL 接入判断

5. 总结

4.1.2 bundle 通过 `-j` 加载后作业端找不到 jar