Flink 系列 18. Flink Table API 编程入门

1. Table API 是什么

Flink 中常见的编程方式主要有三类：

编程方式	特点	适合场景
DataStream API	最底层、最灵活，可以直接控制状态、时间、窗口、算子	复杂流处理、强定制逻辑
Table API	类 SQL 的关系型 API，使用 Java、Scala、Python 代码调用	希望用代码方式编写关系型计算逻辑
SQL	直接编写 SQL	数据开发、实时数仓、ETL、报表统计

Table API 是 Flink 提供的一套统一的关系型 API，可以同时用于流处理和批处理。它的抽象对象是 Table，使用方式和 SQL 比较接近，但是不是直接写 SQL 字符串，而是通过代码链式调用完成查询逻辑。

例如：

Table result = orders
    .select($("user_id"), $("amount"))
    .where($("amount").isGreater(100));

这个逻辑和 SQL 很像：

SELECT user_id, amount
FROM orders
WHERE amount > 100;

Table API 的好处是：

1. 代码中可以获得一定的类型检查和 IDE 提示。
2. 可以和 DataStream API 互相转换。
3. 可以复用 Flink SQL 的优化能力。
4. 可以统一处理流表和批表。

2. 为什么需要 Table API

如果只用 DataStream API，很多关系型处理逻辑会写得比较繁琐。

例如：

• 字段选择
• 条件过滤
• 分组聚合
• 窗口聚合
• 多表关联
• 维表关联

这些逻辑本质上更像 SQL。

Table API 和 SQL 的出现，就是为了让流处理可以用关系型方式来描述。Flink 会把 Table API 或 SQL 转换成底层执行计划，再交给 Flink Runtime 执行。

可以理解成：

Table API / SQL
        |
        v
Flink Planner 优化执行计划
        |
        v
Flink Runtime 执行任务

3. 动态表

学习 Flink Table API，最重要的概念是动态表。

普通数据库中的表是相对静态的，执行一次 SQL 得到一个结果集：

SELECT count(*) FROM orders;

在流处理中，输入数据是不断到来的。例如订单流中每秒都有新订单进来，那么对应的订单表也会一直变化。

Flink 把这种随时间变化的表称为动态表。

动态表可以简单理解为：

Kafka / DataStream 中不断到来的数据
        |
        v
一张不断变化的 Table
        |
        v
持续查询生成新的动态结果表

例如统计每个渠道的订单数：

SELECT channel, count(*)
FROM orders
GROUP BY channel;

在流处理场景中，这个结果不是一次性完成，而是随着新数据到来不断更新。

4. TableEnvironment

TableEnvironment 是 Table API 和 SQL 的入口。

它主要负责：

• 注册表
• 注册 Catalog
• 执行 SQL
• 创建 Table
• 注册函数
• 在 DataStream 和 Table 之间转换

如果只使用 Table API / SQL，可以创建普通 TableEnvironment：

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

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

TableEnvironment tableEnv = TableEnvironment.create(settings);

如果需要和 DataStream API 互相转换，需要使用 StreamTableEnvironment：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();

StreamTableEnvironment tableEnv =
    StreamTableEnvironment.create(env);

实际项目中，如果已经有 DataStream 代码，一般使用 StreamTableEnvironment。

5. 添加依赖

如果是 Maven 项目，通常需要引入 Table API 和 Planner 相关依赖。

下面以 Flink 1.19 为例：

<properties>
    <flink.version>1.19.0</flink.version>
</properties>

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

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner-loader</artifactId>
    <version>${flink.version}</version>
</dependency>

如果要读取 Kafka，还需要 Kafka connector：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>3.2.0-1.19</version>
</dependency>

注意：Flink 版本和 connector 版本需要匹配，不要随意混用。

6. 使用 SQL DDL 创建表

Table API 可以直接使用 SQL DDL 注册表。

例如创建一个 Kafka Source 表：

tableEnv.executeSql(
    "CREATE TABLE orders ("
        + " order_id STRING,"
        + " user_id STRING,"
        + " channel STRING,"
        + " amount DECIMAL(10, 2),"
        + " event_time TIMESTAMP(3),"
        + " WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND"
        + ") WITH ("
        + " 'connector' = 'kafka',"
        + " 'topic' = 'orders',"
        + " 'properties.bootstrap.servers' = 'kafka01:9092',"
        + " 'properties.group.id' = 'flink-table-demo',"
        + " 'scan.startup.mode' = 'latest-offset',"
        + " 'format' = 'json'"
        + ")"
);

这里有几个关键点：

配置	说明
connector	表示数据源类型，例如 kafka、filesystem、jdbc
topic	Kafka topic
properties.bootstrap.servers	Kafka 地址
format	数据格式，例如 json、csv、avro
WATERMARK	事件时间水位线

创建表之后，就可以用 SQL 查询：

Table result = tableEnv.sqlQuery(
    "SELECT channel, amount FROM orders WHERE amount > 100"
);

也可以直接执行 SQL：

tableEnv.executeSql(
    "SELECT channel, amount FROM orders WHERE amount > 100"
).print();

7. 使用 Table API 查询

如果使用 Table API，可以先获取表对象：

import static org.apache.flink.table.api.Expressions.$;

Table orders = tableEnv.from("orders");

然后进行查询：

Table result = orders
    .select($("order_id"), $("user_id"), $("channel"), $("amount"))
    .where($("amount").isGreater(100));

这段代码等价于：

SELECT order_id, user_id, channel, amount
FROM orders
WHERE amount > 100;

如果要打印结果：

result.execute().print();

Table API 的常见操作如下：

操作	Table API	SQL
查询字段	select	SELECT
过滤	where / filter	WHERE
分组	groupBy	GROUP BY
聚合	sum / count / avg	SUM / COUNT / AVG
排序	orderBy	ORDER BY
关联	join	JOIN

8. DataStream 转 Table

如果数据已经在 DataStream 中，可以转换为 Table。

先定义数据对象：

public class OrderEvent {
    public String orderId;
    public String userId;
    public String channel;
    public Double amount;
    public Long ts;
}

DataStream 转 Table：

DataStream<OrderEvent> orderStream = env
    .fromSource(source, watermarkStrategy, "order-source");

Table orderTable = tableEnv.fromDataStream(
    orderStream,
    $("orderId"),
    $("userId"),
    $("channel"),
    $("amount"),
    $("ts")
);

如果要注册成临时视图：

tableEnv.createTemporaryView("orders_from_stream", orderTable);

然后就可以使用 SQL：

Table result = tableEnv.sqlQuery(
    "SELECT channel, count(*) AS cnt "
        + "FROM orders_from_stream "
        + "GROUP BY channel"
);

9. Table 转 DataStream

Table 也可以转回 DataStream。

如果结果只会追加，不会更新，可以使用 append stream 的思路：

DataStream<Row> resultStream = tableEnv.toDataStream(result);

如果结果会更新，例如 GROUP BY 聚合产生更新流，需要用 changelog stream：

DataStream<Row> changelogStream = tableEnv.toChangelogStream(result);

这点非常重要。

例如下面这种查询：

SELECT channel, count(*)
FROM orders
GROUP BY channel;

它的结果会随着新订单不断更新。对于某个 channel 来说，count 会从 1 变成 2，再变成 3。这个结果不是简单追加，而是更新。

所以在老版本 Flink 中常见 toRetractStream，在较新版本中更推荐理解为 changelog：

INSERT：新增结果
UPDATE_BEFORE：更新前结果
UPDATE_AFTER：更新后结果
DELETE：删除结果

10. 窗口聚合

流处理中经常需要按时间窗口统计数据。

例如每 10 秒统计一次不同渠道的订单数。

在 Flink SQL 中可以这样写：

SELECT
    window_start,
    window_end,
    channel,
    COUNT(*) AS cnt
FROM TABLE(
    TUMBLE(TABLE orders, DESCRIPTOR(event_time), INTERVAL '10' SECOND)
)
GROUP BY window_start, window_end, channel;

这里使用的是窗口表值函数，核心是：

TUMBLE(TABLE orders, DESCRIPTOR(event_time), INTERVAL '10' SECOND)

含义是：

参数	说明
orders	输入表
event_time	事件时间字段
INTERVAL '10' SECOND	10 秒滚动窗口

如果用 Table API，可以这样表达：

Table result = orders
    .window(Tumble.over(lit(10).seconds()).on($("event_time")).as("w"))
    .groupBy($("channel"), $("w"))
    .select(
        $("channel"),
        $("w").start().as("window_start"),
        $("w").end().as("window_end"),
        $("order_id").count().as("cnt")
    );

窗口聚合要注意两点：

1. 必须有时间字段。
2. 如果使用事件时间，必须定义 watermark。

11. 事件时间和 Watermark

流处理中的时间通常分为三类：

时间类型	说明
Processing Time	数据被 Flink 处理时的机器时间
Event Time	数据真实发生时间
Ingestion Time	数据进入 Flink 的时间

实时计算中最常用的是 Event Time。

因为数据可能乱序到达，例如：

10:00:01 的订单先到
10:00:00 的订单后到

如果只按处理时间计算，统计结果就可能不准确。

在 Table API / SQL 中，可以在建表时定义 watermark：

event_time TIMESTAMP(3),
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND

含义是允许数据最多乱序 5 秒。

如果超过 watermark 之后才到达，数据就会被认为是迟到数据。

12. Table API 和 SQL 的关系

Table API 和 SQL 最终都会交给 Flink Planner 处理。

可以简单理解为：

Table API
    |
    v
Relational Plan
    ^
    |
SQL

所以很多场景中，Table API 和 SQL 可以互相替换。

选择时可以参考：

场景	推荐方式
数据开发人员写统计逻辑	SQL
Java 项目中需要编程式组合查询	Table API
需要和 DataStream 深度结合	StreamTableEnvironment + Table API
复杂状态逻辑、定制算子	DataStream API

我的建议是：入门时先掌握 SQL，再理解 Table API。因为 SQL 更直观，而 Table API 的底层概念和 SQL 是一致的。

13. 一个完整示例

下面用一个简单例子串起来。

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class FlinkTableApiDemo {

    public static void main(String[] args) {
        EnvironmentSettings settings = EnvironmentSettings
            .newInstance()
            .inStreamingMode()
            .build();

        TableEnvironment tableEnv = TableEnvironment.create(settings);

        tableEnv.executeSql(
            "CREATE TABLE orders ("
                + " order_id STRING,"
                + " user_id STRING,"
                + " channel STRING,"
                + " amount DECIMAL(10, 2),"
                + " event_time TIMESTAMP(3),"
                + " WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND"
                + ") WITH ("
                + " 'connector' = 'datagen',"
                + " 'rows-per-second' = '5',"
                + " 'fields.channel.length' = '3'"
                + ")"
        );

        Table result = tableEnv.sqlQuery(
            "SELECT channel, COUNT(*) AS cnt "
                + "FROM orders "
                + "GROUP BY channel"
        );

        result.execute().print();
    }
}

这个例子使用 datagen 生成测试数据，不依赖 Kafka，适合作为本地入门测试。

运行后可以看到类似结果：

+I[abc, 1]
-U[abc, 1]
+U[abc, 2]
-U[abc, 2]
+U[abc, 3]

这里的 +I、-U、+U 就是 changelog 语义：

标识	含义
+I	Insert
-U	Update Before
+U	Update After
-D	Delete

这也是 Table API / SQL 做流式聚合时必须理解的地方。

14. 和旧版本写法的区别

原来很多 Flink 1.7 左右的示例会这样写：

val tableEnv: StreamTableEnvironment =
  TableEnvironment.getTableEnvironment(env)

或者：

val resultDstream = resultTable.toRetractStream[(String, Long)]

这类写法在老版本中很常见，但如果按照 2024 年常见的 Flink 1.18、1.19 项目来看，更推荐使用：

StreamTableEnvironment.create(env)

以及：

tableEnv.toDataStream(table)
tableEnv.toChangelogStream(table)

所以整理这篇文章时，没有继续使用老版本的 'field、toRetractStream 作为主线，而是改成了当前更容易迁移的写法。

15. 总结

Flink Table API 的核心可以总结成几句话：

1. Table API 是 Flink 的关系型编程接口。
2. Table API 和 SQL 都基于动态表。
3. 流上的查询不是一次性结果，而是持续更新的结果。
4. Append 结果可以直接转 DataStream。
5. 聚合、Join 等会更新的结果需要理解 changelog。
6. 事件时间窗口必须定义时间字段和 watermark。

如果刚开始学习 Flink Table API，不建议一开始就写复杂代码，可以按下面顺序学习：

TableEnvironment
SQL DDL 建表
Table API select / where
DataStream 和 Table 互转
动态表和 changelog
窗口聚合
Kafka / JDBC / Filesystem Connector

把这些概念理清楚之后，再看 Flink SQL 实时数仓、维表 Join、窗口统计、CDC 入湖等内容就会顺很多。