4. Table API & SQL 统一层：动态表与流-表转换

好，我们进入第四章。这一章可以说是整个流批一体架构的「翻译官」——动态表（Dynamic Table）。说白了，它就是让 SQL 这种静态查询语言，能理解实时流数据的核心机制。

我记得刚接触 Flink 那会儿，总觉得「流就是流，表就是表」，两个东西怎么能统一呢？后来才明白，动态表就是那个桥梁。它把无界的数据流，变成了一个有界的、不断变化的「视图」。你想想看，批处理查的是静态表，流处理查的是动态表——本质上都是查表，只是后者会随时间变化。

4.1 动态表的核心概念

动态表，顾名思义，就是会随时间变化的表。在批处理中，表是静止的，查询一次就结束了。但在流处理中，数据源源不断进来，表的内容也在持续更新。

举个例子。假设你有一个订单流，每秒都有新订单进来。在批处理中，你得等一天结束才能查今天的订单总额。但在动态表上，你可以随时查询——每次查询看到的是当前时刻的快照，下一秒可能就变了。

我个人习惯把动态表理解成「活的视图」。它不像传统数据库视图那样被动刷新，而是主动接收变更事件，实时更新结果。

4.2 流-表转换的三种模式

流怎么变成表？表怎么变回流？这里涉及三种转换模式：Append、Retract、Upsert。我在项目中踩过不少坑，尤其是 Retract 和 Upsert 的区别，很多人一开始都搞混。

4.2.1 Append 模式

Append 是最简单的模式。它只追加新行，不修改已有数据。适合日志、点击流这类只增不减的场景。

-- 流 -> 动态表（Append）
CREATE TABLE click_events (
  user_id STRING,
  page_url STRING,
  click_time TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'topic' = 'clicks',
  'format' = 'json'
);

-- 查询：统计每个用户的点击次数
SELECT user_id, COUNT(*) AS cnt
FROM click_events
GROUP BY user_id;

这里要注意：虽然源表是 Append 模式，但 GROUP BY 之后的结果表会变成 Retract 或 Upsert 模式——因为聚合结果会随着新数据到来而更新。

4.2.2 Retract 模式

Retract 模式用「撤回+新增」的方式表示更新。它发出两条消息：一条是旧值的撤回（-），一条是新值的插入（+）。

-- 动态表 -> 流（Retract）
-- 假设有一个按小时统计的订单金额表
-- 当新数据到来，旧的小时聚合结果被撤回，新的被插入

-- 输出流中会看到：
-- + (2024-01-01 10:00, 100.0)  -- 插入
-- - (2024-01-01 10:00, 100.0)  -- 撤回旧值
-- + (2024-01-01 10:00, 150.0)  -- 插入新值

4.2.3 Upsert 模式

Upsert 模式更高效。它直接根据主键更新或插入，不需要先撤回再插入。适合有明确主键的场景，比如数据库的变更数据捕获（CDC）。

-- 动态表 -> 流（Upsert）
CREATE TABLE user_profile (
  user_id STRING PRIMARY KEY NOT ENFORCED,
  name STRING,
  age INT
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'user-profile',
  'key.format' = 'json',
  'value.format' = 'json'
);

-- 更新用户年龄
INSERT INTO user_profile
SELECT user_id, name, age FROM user_changes;

4.3 三种模式对比

模式	消息数量	是否支持更新	是否需要主键	典型场景
Append	1 条（+）	否	否	日志、点击流、事件日志
Retract	2 条（- 和 +）	是	否	无主键的聚合结果、窗口计算
Upsert	1 条（+ 或 更新）	是	是	CDC 数据、维表、状态更新

4.4 CDC 数据接入

CDC（Change Data Capture）是流批一体架构中非常重要的数据源。它实时捕获数据库的变更（插入、更新、删除），然后以流的形式接入 Flink。

我做过一个实时数仓项目，业务库是 MySQL，每天几亿条变更。用 Debezium + Kafka 把 CDC 数据接入 Flink，然后做实时 ETL。这里的关键是如何正确处理 CDC 中的删除事件。

-- 使用 Flink CDC 连接器接入 MySQL 数据
CREATE TABLE orders_cdc (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_id BIGINT,
  product_id BIGINT,
  amount DECIMAL(10, 2),
  order_time TIMESTAMP(3),
  -- CDC 元数据字段
  op_type STRING METADATA FROM 'op_type'  -- 'c'=创建, 'u'=更新, 'd'=删除
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'database-name' = 'shop',
  'table-name' = 'orders',
  'server-id' = '5401'
);

-- 实时统计每个用户的订单总额（自动处理删除）
SELECT user_id, SUM(amount) AS total_amount
FROM orders_cdc
GROUP BY user_id;

4.5 维表 Join 与异步 IO

维表 Join 是实时计算中的常见需求。比如订单流需要关联用户维表，获取用户的地理位置、会员等级等信息。

但维表 Join 有个性能瓶颈：如果每条流数据都去查一次外部存储（比如 Redis、MySQL），延迟会很高。这时候就需要异步 IO 来优化。

4.5.1 维表 Join 的三种方式

方式	延迟	一致性	适用场景
同步查询	高（每次请求等待响应）	强	数据量小、延迟不敏感
缓存 + 同步	中（缓存命中时低）	最终一致	维表变化不频繁
异步 IO	低（并发请求）	强	高吞吐、低延迟要求

4.5.2 异步 IO 实战

异步 IO 的核心思想是：不等待单个请求返回，而是批量发送请求，然后统一处理响应。Flink 的 AsyncDataStream 提供了现成的支持。

// 异步查询 Redis 维表
public class AsyncRedisLookup extends RichAsyncFunction<Order, EnrichedOrder> {
    private transient JedisPool jedisPool;

    @Override
    public void open(Configuration parameters) {
        jedisPool = new JedisPool("localhost", 6379);
    }

    @Override
    public void asyncInvoke(Order order, ResultFuture<EnrichedOrder> resultFuture) {
        // 异步查询用户信息
        CompletableFuture.supplyAsync(() -> {
            try (Jedis jedis = jedisPool.getResource()) {
                String userInfo = jedis.get("user:" + order.userId);
                return parseUserInfo(userInfo);
            }
        }).thenAccept(userInfo -> {
            resultFuture.complete(
                Collections.singletonList(new EnrichedOrder(order, userInfo))
            );
        });
    }
}

// 在 DataStream API 中使用
DataStream<EnrichedOrder> enrichedStream = 
    AsyncDataStream.unorderedWait(
        orderStream,
        new AsyncRedisLookup(),
        1000,  // 超时时间
        TimeUnit.MILLISECONDS,
        100    // 最大并发请求数
    );

4.6 动态表转换全景图

下面这张图展示了流、动态表、SQL 查询之间的完整转换关系。你可以把它当作本章的「地图」。

4.7 避坑指南与最佳实践

最后，分享几个我在实战中总结的经验：

• 状态大小控制：动态表的状态会随着时间增长。如果做 GROUP BY user_id，每个用户都要维护一个状态。我见过一个项目，用户量上亿，状态直接撑爆了 TaskManager 内存。解决方案是加时间窗口，或者用 RocksDB 状态后端。
• 事件时间 vs 处理时间：动态表默认使用事件时间。如果数据乱序严重，记得设置 watermark 和 allowedLateness。我曾经因为没设 watermark，导致窗口结果一直不触发，排查了半天才发现是时间戳问题。
• 维表缓存策略：如果维表数据变化不频繁（比如每天更新一次），建议用 全量缓存，定时刷新。如果变化频繁，用 LRU 缓存 + 异步 IO。别用同步查询，延迟会高到让你怀疑人生。
• CDC 数据的一致性：CDC 数据可能重复或乱序。Flink 的 CDC 连接器会处理大部分情况，但如果你自己写连接器，记得实现 幂等写入 或 去重逻辑。