📚 课程目录第3章 Flink SQL基础回顾

3. Flink SQL 基础回顾：Table & SQL API、动态表概念、CDC 与 Changelog Stream

好，咱们进入第三章。这一章算是承上启下——如果你对 Flink SQL 已经很熟了，可以快速扫一遍；如果刚接触，那这部分就是后面所有实战的基石。

我个人习惯把 Flink SQL 看作「流批一体」的入口。你写一条 SQL，Flink 自己判断是跑在流上还是跑在批上。听起来很美好，但实际用起来有几个关键概念必须掰扯清楚。

3.1 Table & SQL API：两种写法，一个内核

Flink 提供了两套关系型 API：Table API 和 SQL API。Table API 是嵌入在 Java/Scala 里的 DSL，SQL API 就是纯 SQL 字符串。底层共享同一套优化器和执行引擎。

对比维度	Table API	SQL API
表达方式	Java/Scala 链式调用	纯 SQL 字符串
类型安全	编译期检查	运行时解析
复杂逻辑	容易嵌套条件	需要 UDF 辅助
运维友好	需重新编译	可动态下发
我常用的场景	ETL 中做多路输出	即席查询、CDC 入湖

举个例子，用 Table API 注册一个数据源：

Table sourceTable = tableEnv.fromDataStream(
    dataStream,
    Schema.newBuilder()
        .column("id", DataTypes.INT())
        .column("name", DataTypes.STRING())
        .column("ts", DataTypes.TIMESTAMP(3))
        .watermark("ts", "ts - INTERVAL '5' SECOND")
        .build()
);

换成 SQL 就是：

tableEnv.executeSql(
    "CREATE TABLE source_table (" +
    "  id INT, " +
    "  name STRING, " +
    "  ts TIMESTAMP(3), " +
    "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND" +
    ") WITH (...)"
);

我的习惯：开发阶段用 Table API 方便调试，上线前改成 SQL 放到运维平台。这样既享受了 IDE 的代码补全，又保留了动态修改的灵活性。

3.2 动态表：流上的关系视图

传统 SQL 处理的是静态表——数据不动查询动。但流处理是数据一直在动，查询也一直在跑。Flink 用 动态表 来桥接这个矛盾。

说白了，动态表就是随时间不断变化的表。每个时间点你查它，结果都不一样。Flink 把流转换成动态表，然后对动态表执行连续查询，查询结果又是一个动态表，最后再转回流输出。

我画了一张图帮你理解这个流程：

这里有个容易踩的坑——不是所有 SQL 都能在流上直接跑。比如无限制的 GROUP BY，状态会无限膨胀。Flink 通过状态 TTL 和窗口来约束，后面章节会细讲。

3.3 CDC 与 Changelog Stream：捕获变化，传递变化

CDC（Change Data Capture）这几年特别火。说白了就是监听数据库的 binlog，把每一行数据的 INSERT、UPDATE、DELETE 操作实时同步出来。Flink 对 CDC 的支持非常成熟，尤其是和 Iceberg 配合时，简直是绝配。

Flink 内部用 Changelog Stream 来表示这种带变更类型的数据流。每条记录除了业务字段，还带一个 RowKind 标记：

RowKind	含义	对应数据库操作
`INSERT`	新增一行	INSERT
`UPDATE_BEFORE`	更新前的旧值	UPDATE（前镜像）
`UPDATE_AFTER`	更新后的新值	UPDATE（后镜像）
`DELETE`	删除一行	DELETE

  重点：Iceberg 的 MOR（Merge-On-Read）模式天然支持 CDC 入湖。Flink 把 Changelog Stream 写入 Iceberg 时，INSERT 变成新文件，UPDATE/DELETE 变成 delete 文件。读的时候合并，写的时候不阻塞。

我记得第一次在生产上跑 CDC 入湖时，遇到一个坑——MySQL CDC 源表默认只发 UPDATE_AFTER，不发 UPDATE_BEFORE。这导致 Iceberg 的 delete 文件无法正确标记旧数据。解决方案是在 Flink CDC connector 里配置 'scan.incremental.snapshot.enabled' = 'false' 或者显式要求发送前镜像。

看一段实际代码，用 Flink SQL 定义 CDC 源表：

CREATE TABLE mysql_cdc_source (
    id INT,
    name STRING,
    age INT,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '192.168.1.100',
    'port' = '3306',
    'username' = 'cdc_user',
    'password' = 'cdc_pass',
    'database-name' = 'test_db',
    'table-name' = 'user_table',
    'scan.incremental.snapshot.enabled' = 'false',
    'debezium.snapshot.mode' = 'initial'
);

然后直接查这个表，Flink 会自动把 binlog 转成 Changelog Stream：

-- 查看实时变更
SELECT id, name, age FROM mysql_cdc_source;

-- 或者做聚合
SELECT
    COUNT(*) AS user_count,
    TUMBLE_END(proc_time, INTERVAL '1' MINUTE) AS window_end
FROM mysql_cdc_source
GROUP BY TUMBLE(proc_time, INTERVAL '1' MINUTE);

注意：CDC 源表在 Flink 中默认是 动态表，不是静态表。你写 SELECT * 会一直输出变更事件，不会结束。这在做批流混合查询时容易混淆——我见过有人用 CDC 表做维表 JOIN，结果状态越积越大，最后 OOM。

3.4 三种模式：Append、Retract、Upsert

Flink 的动态表查询结果，根据 SQL 类型不同，会输出三种 Changelog 模式：

Append 模式：只有 INSERT，没有 UPDATE/DELETE。比如简单的 SELECT * FROM source，或者无重复键的窗口聚合。
Retract 模式：用 INSERT 表示新增，用 DELETE 表示撤回。比如 GROUP BY 后的 COUNT，新数据来的时候要先撤回旧结果，再发新结果。
Upsert 模式：用 UPDATE_BEFORE + UPDATE_AFTER 表示更新，DELETE 表示删除。需要定义 PRIMARY KEY，比如 CDC 源表或者带 key 的聚合。

这三种模式直接影响下游 Iceberg 的写入策略。Append 模式直接写数据文件；Retract 和 Upsert 需要写 delete 文件。选错了模式，轻则数据重复，重则查询结果错误。

我建议你在写 Flink SQL 之前，先想清楚：这条 SQL 的输出是 append-only 还是带更新的？ 如果是带更新的，必须确保下游 sink 支持 upsert 或 retract。Iceberg 支持，但需要正确配置。

嗯，这一章的基础概念就这些。后面我们写 Iceberg 实时入湖作业时，会反复用到动态表和 Changelog Stream 这两个概念。理解了它们，Flink + Iceberg 的配合就清晰了。