4. Flink Table & SQL：从API到SQL的流批统一之路

这一章我们来聊聊 Flink 的 Table API 和 SQL。说实话，很多初学者会问：既然有了 DataStream API，为什么还要搞一套 Table & SQL？我的回答很简单——降低门槛，提升效率。

我在 2019 年接手过一个实时数仓项目，团队里一半是 Java 工程师，一半是数据分析师。让数据分析师去写 DataStream 的算子？那简直是灾难。但换成 SQL，他们半天就能上手。这就是 Table & SQL 的价值所在。

4.1 Table API 与 SQL：同根同源，不同用法

Table API 和 SQL，本质上是一回事。它们共享同一个底层引擎——Flink 的查询优化器。你写的 Table API 代码，最终会被翻译成 SQL 的逻辑计划，然后走同样的优化和执行路径。

那区别在哪？说白了就是表达方式不同：

• SQL：声明式，你告诉系统“要什么”，系统决定“怎么做”
• Table API：嵌入式 DSL，在 Java/Scala 代码里链式调用，适合复杂逻辑拼接

我个人习惯这样划分：

场景	推荐方式	原因
简单查询、聚合	SQL	简洁直观，非开发人员也能看懂
多步骤ETL、条件分支	Table API	代码内可加逻辑判断，灵活拼接
与DataStream互操作	Table API	toDataStream / fromDataStream 原生支持
动态表定义	两者皆可	SQL注册临时表，API用TableEnvironment

来看一段对比代码。同样的功能——从 Kafka 读取订单流，按用户ID分组统计金额：

// SQL 方式
Table orders = tableEnv.sqlQuery(
    "SELECT user_id, SUM(amount) AS total_amount " +
    "FROM KafkaOrders " +
    "GROUP BY user_id"
);

// Table API 方式
Table orders = tableEnv.from("KafkaOrders")
    .groupBy($("user_id"))
    .select($("user_id"), $("amount").sum().as("total_amount"));

你想想看，如果后面要加过滤条件、窗口定义，SQL 需要改字符串，Table API 直接链式调用就行。这就是灵活性的差异。

4.2 动态表与连续查询：流式SQL的核心机制

这是整个流式 SQL 的基石。不理解动态表，你就没法真正用好 Flink SQL。

动态表是什么？简单说，就是一张会随时间变化的关系表。传统数据库的表是静态的，你查的时候数据不动。但流式场景下，数据源源不断进来，表的内容一直在变。

连续查询呢？就是持续运行在动态表上的查询。每次动态表有变化，查询结果就自动更新。

我画了一张图，帮你理解这个关系：

这里有个关键点：动态表的更新模式有三种：

• Append-only：只追加，不修改。比如点击日志流
• Retract：先发一条撤回消息，再发一条新消息。比如 GROUP BY 聚合结果变化时
• Upsert：根据 Key 更新或插入。比如维表 JOIN 后的结果

4.3 流式SQL的局限性：不是所有SQL都能跑在流上

嗯，这里要泼一盆冷水。流式 SQL 虽然强大，但并不是万能的。有些在批处理里很自然的操作，到了流式场景就变得棘手甚至不可行。

第一个大坑：完整的 ORDER BY

批处理里，SELECT * FROM table ORDER BY ts 没问题。但在流上，数据是无限的，你永远不知道下一个数据会不会比当前的小。所以 Flink 只支持带 LIMIT 的排序，或者窗口内的排序。

-- 这个可以：窗口内的 Top-N
SELECT * FROM (
    SELECT *, ROW_NUMBER() OVER (
        PARTITION BY window_end
        ORDER BY amount DESC
    ) AS rn FROM TumblingOrders
) WHERE rn <= 3;

-- 这个不行：全局排序
SELECT * FROM KafkaOrders ORDER BY ts;  -- ❌ 会报错

第二个大坑：非等值 JOIN

流式 JOIN 需要维护状态。等值 JOIN 可以用哈希分区，状态可控。但非等值 JOIN（比如 a.ts BETWEEN b.start_ts AND b.end_ts）需要笛卡尔积式的匹配，状态会爆炸。

第三个大坑：多表 JOIN 的乱序问题

流式 JOIN 依赖 watermark 来判断数据是否到齐。如果 watermark 设置不合理，或者数据乱序严重，JOIN 结果可能不准确。我建议：

• 给每个流设置合理的 watermark 延迟
• 使用 table.exec.source.idle-timeout 处理空闲源
• 对延迟数据做侧输出流兜底

4.4 批流统一的SQL语法：一套SQL，两种语义

这是 Flink 1.12 之后最重要的特性之一。你写同样的 SQL，在批模式和流模式下都能跑。区别只在于执行引擎的行为：

特性	批模式	流模式
数据源	有界数据集（文件、Hive表）	无界数据流（Kafka、Pulsar）
GROUP BY	一次性计算，结果完整	持续更新，结果随数据变化
JOIN	全量关联，结果确定	基于窗口或状态，结果渐进
ORDER BY	全局排序，完整输出	仅支持窗口内排序或Top-N
状态管理	不需要状态	需要状态，可配置TTL

你想想看，这意味着什么？意味着你可以用同一套 SQL 脚本，在开发环境用批模式快速验证逻辑，上线后切到流模式跑实时数据。逻辑完全一致，只是执行方式不同。

来看一个实际例子。假设我们要统计每分钟的订单金额：

-- 这段SQL在批和流模式下都能跑
INSERT INTO ResultTable
SELECT
    TUMBLE_START(ts, INTERVAL '1' MINUTE) AS window_start,
    TUMBLE_END(ts, INTERVAL '1' MINUTE) AS window_end,
    user_id,
    SUM(amount) AS total_amount
FROM KafkaOrders
GROUP BY
    TUMBLE(ts, INTERVAL '1' MINUTE),
    user_id;

批模式下，Flink 会一次性读取所有数据，计算窗口聚合，输出完整结果。流模式下，Flink 会持续接收数据，每个窗口结束时输出一次结果。

我个人习惯在开发阶段这样做：

1. 先用批模式 + 小数据集验证 SQL 逻辑正确性
2. 再用流模式 + 历史数据回放测试实时性
3. 最后上线流模式，设置好 checkpoint 和状态 TTL

这样一套流程下来，SQL 的调试效率能提升不少。你想想看，如果每次改 SQL 都要部署一个流任务去验证，那得多浪费时间？

最后提一句，批流统一的 SQL 语法，背后依赖的是 Flink 的 Blink Planner。它会在编译阶段根据运行模式，生成不同的物理执行计划。批模式走批式算子，流模式走流式算子。这个细节我们后面章节再展开。