📚 课程目录第4章 Flink SQL与Table API

4、Flink SQL与Table API：动态表、连续查询与CDC入湖入仓

说实话，Flink 的 Table API 和 SQL 是我这几年用得最多的接口。很多人觉得「SQL 嘛，谁不会写」，但在流处理里，事情没那么简单。

你想想看，传统数据库里一张表，数据是静止的。你查一次，结果就固定了。但在 Flink 里，数据一直在流，表的内容每时每刻都在变。那 SQL 怎么查？查出来的结果又怎么更新？

这就是 Flink 动态表要解决的核心问题。我当年第一次接触这个概念时，也绕了好一阵子。今天咱们就把这块彻底讲透。

4.1 动态表：流上的「快照」

动态表，说白了就是一张会随时间变化的关系表。每条新数据进来，表的内容就更新一次。你可以把它理解成数据库里的物化视图——只不过这个视图是实时刷新的。

核心要点：动态表是流与 SQL 之间的桥梁。流转换成动态表，SQL 查询动态表，结果再转换成新的流。

举个例子。假设你有一个订单流：

// 订单流样例
{user: "Alice", product: "手机", amount: 5000, time: "2024-01-01 10:00"}
{user: "Bob",   product: "电脑", amount: 8000, time: "2024-01-01 10:01"}
{user: "Alice", product: "耳机", amount: 200,  time: "2024-01-01 10:02"}

这个流转换成动态表后，每来一条记录，表里就多一行。但注意——动态表不存储所有历史数据，它只保留当前「可见」的数据窗口。

我个人习惯把动态表想象成「河流上的浮标」。河水在流，浮标的位置一直在变。你每次看，浮标都在新位置。动态表也一样，每次查询，结果都可能不同。

4.2 连续查询：一直在跑的 SQL

普通 SQL 查一次就结束了。连续查询不一样——它一直在跑，输入数据一变，输出结果就跟着变。

我举个例子你就明白了：

-- 连续查询示例：统计每个用户的订单总额
SELECT 
  user, 
  SUM(amount) AS total_amount
FROM orders
GROUP BY user;

这条 SQL 在传统数据库里跑一次，给你一个结果。但在 Flink 里，它会一直运行。Alice 每下一单，她的 total_amount 就更新一次。

这里有个坑，我曾经踩过——连续查询的结果也是动态表。也就是说，查询结果本身也在不断变化。那怎么把结果输出到外部系统？

Flink 提供了三种更新模式：

更新模式	说明	适用场景
Append 模式	只追加，不更新已有行	日志、事件流
Retract 模式	先发送一条删除消息，再发送一条插入消息	聚合查询、去重
Upsert 模式	根据 key 直接更新或插入	维表关联、CDC 输出

注意：不是所有 SQL 都支持所有模式。比如 GROUP BY 产生的更新，只能用 Retract 或 Upsert 模式。如果你用 Append 模式去接聚合结果，Flink 会直接报错。

4.3 维表 Join：流处理中的「查字典」

流处理里最头疼的问题之一，就是维表 Join。你想想看，订单流里只有 user_id，没有用户姓名、地址这些信息。你得去 MySQL 或者 Redis 里查。

但问题来了——流是无限的，维表数据是有限的。而且维表数据还会变。怎么保证 Join 的结果是对的？

Flink 提供了几种维表 Join 的方式：

预加载维表：启动时一次性加载到内存。适合小维表，比如几百条配置数据。
热缓存维表：用 Guava Cache 或 Caffeine，设置过期时间。适合中等规模维表。
异步 IO 查询：每条数据都去查外部存储。适合大维表，但延迟较高。

我项目中用得最多的是热缓存维表。给你看一段代码：

-- 维表定义：从 MySQL 读取用户信息
CREATE TEMPORARY TABLE user_dim (
  user_id   STRING,
  user_name STRING,
  city      STRING,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url'      = 'jdbc:mysql://localhost:3306/db',
  'table-name' = 'users',
  'lookup.cache' = 'PARTIAL',   -- 热缓存模式
  'lookup.cache.max-rows' = '10000',
  'lookup.cache.ttl' = '10 minute'
);

-- 流表与维表 Join
SELECT 
  o.order_id,
  o.amount,
  u.user_name,
  u.city
FROM orders AS o
JOIN user_dim FOR SYSTEM_TIME AS OF o.proc_time AS u
ON o.user_id = u.user_id;

这里有个关键点——FOR SYSTEM_TIME AS OF。它告诉 Flink：在数据到达的那一刻，去查维表当时的值。这样即使维表数据变了，每条数据也能拿到它「应该」看到的版本。

我的经验：维表缓存时间别设太长。我曾经设了 30 分钟，结果用户改了地址，订单还是发到老地址去了。一般 5-10 分钟比较合理。如果对实时性要求极高，就用异步 IO 直查。

4.4 CDC 数据入湖入仓：从 Binlog 到数仓

CDC（Change Data Capture）是这几年最火的数据同步方式。说白了，就是监听数据库的 Binlog，把增删改操作实时同步到数仓或数据湖。

Flink 对 CDC 的支持非常成熟。你只需要一个 Flink CDC connector，就能把 MySQL、PostgreSQL、MongoDB 的数据实时同步到 Hive、Iceberg、Paimon 或者 Kafka。

我画了一张图，帮你理解 CDC 入湖入仓的整体流程：

你看，Flink CDC 在中间扮演了「翻译官」的角色。它把数据库的 Binlog 翻译成 Flink 能理解的 Changelog 流，然后你可以用 SQL 随意处理——过滤、聚合、Join，最后写到任何地方。

给你看一个完整的 CDC 入湖示例：

-- 1. 定义 CDC 源表（监听 MySQL 的 orders 表）
CREATE TABLE cdc_orders (
  order_id   INT,
  user_id    INT,
  amount     DECIMAL(10,2),
  order_time TIMESTAMP(3),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',
  'hostname'  = 'localhost',
  'port'      = '3306',
  'username'  = 'root',
  'password'  = '123456',
  'database-name' = 'mydb',
  'table-name'    = 'orders',
  'scan.startup.mode' = 'latest-offset'  -- 只从当前开始同步
);

-- 2. 定义 Iceberg 目标表（数据湖）
CREATE TABLE iceberg_orders (
  order_id   INT,
  user_id    INT,
  amount     DECIMAL(10,2),
  order_time TIMESTAMP(3),
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
  'connector' = 'iceberg',
  'catalog-database' = 'default',
  'catalog-table'    = 'orders',
  'format-version'   = '2',
  'write.upsert.enabled' = 'true'
);

-- 3. 启动同步作业
INSERT INTO iceberg_orders
SELECT * FROM cdc_orders;

避坑指南：我曾经在生产环境遇到过一个问题——CDC 作业重启后，数据重复了。原因是 scan.startup.mode 设成了 initial，每次重启都从 Binlog 开头读。后来改成 latest-offset 配合 checkpoint，才解决了。记住：CDC 作业一定要开启 checkpoint，否则 Exactly-Once 就是空话。

4.5 动态表与 CDC 的「化学反应」

动态表和 CDC 结合起来，能实现一个非常优雅的架构——实时数仓。

你想想看：

CDC 把数据库的变更实时同步到 Kafka
Flink SQL 消费 Kafka，用动态表做实时 ETL
连续查询的结果写入数据湖或数据仓库

整个过程全是 SQL，不需要写一行 Java 代码。我去年帮一个电商客户做实时数仓，就是用这套方案。原来他们用 Spark 批处理，每天凌晨跑一次，数据延迟 12 小时。换成 Flink SQL 后，延迟降到 5 秒以内。

嗯，这里要注意——不是所有场景都适合 CDC。如果你的源表有大量的 DDL 操作（比如加列、改类型），CDC 作业可能会挂。Flink CDC 2.4 之后支持了 schema evolution，但还不够成熟。我建议你在生产环境先做充分的压力测试。

我的建议：如果你刚开始接触 Flink SQL，先从简单的 Append 流开始。比如把 Kafka 数据写到 HDFS，不要一上来就搞 CDC + 维表 Join + 聚合。等把动态表的概念吃透了，再逐步加复杂度。

动态表、连续查询、维表 Join、CDC 入湖入仓——这四个概念是 Flink SQL 的基石。理解了它们，你就能用 SQL 写出以前需要几百行 Java 代码才能实现的流处理逻辑。说白了，Flink 把流处理的复杂度藏在了 SQL 引擎里，让你用最熟悉的方式解决最复杂的问题。