Flink SQL Query 语法（一）

Posted 2023-04-05

参考技术A

SELECT 语句和 VALUES 语句需要使用 TableEnvironment 的 sqlQuery() 方法加以指定，会以 Table 的形式返回 SELECT （或 VALUE）的查询结果。Table 可被用于 SQL 或 Table API 查询、转换为 DataSet 或 DataStream、输出到 TableSink。SQL 与 Table API 的查询可以进行无缝融合、整体优化。

为了可以在 SQL 查询中访问到表，需要先在 TableEnvironment 中 注册表 （可以通过 TableSource、Table、CREATE TABLE 语句、DataStream 或 DataSet 注册）。为方便起见 Table.toString() 将会在其 TableEnvironment 中以唯一的名称自动注册表，并返回名称。

注意： 查询若包括了不支持的 SQL 特性，将会抛出 TableException 。

以下示例显示如何在已注册和内联表上指定 SQL 查询。

SELECT 语句或者 VALUES 语句可以通过 TableEnvironment.executeSql() 方法来执行，该方法返回 TableResult 对象用于包装查询的结果，一个 Table 对象可以通过 Table.execute() 方法执行获取查询结果。 TableResult.collect() 方法返回一个可以关闭的行迭代器（除非所有的数据都被收集到本地，否则一个查询作业永远不会结束。所以通过 CloseableIterator#close() 方法主动地关闭作业以防止资源泄露）。 还可以通过 TableResult.print() 方法将查询结果打印到控制台。TableResult 中的结果数据只能被访问一次，因此一个 TableResult 实例中， collect() 方法和 print() 方法不能被同时使用。

TableResult.collect() 与 TableResult.print() 的行为在不同的 checkpointing 模式下略有不同。

Flink 通过支持标准 ANSI SQL的 Apache Calcite 解析 SQL。以下“BNF-语法”描述了批处理和流处理查询中所支持的 SQL 特性的超集。

Flink SQL 对于标识符（表、属性、函数名）的命名策略类似于 Java 的词法约定：

字符串文本常量需要被单引号包起来（如 SELECT \'Hello World\' ）。两个单引号表示转义（如 SELECT \'It\'\'s me.\' ）。字符串文本常量支持 Unicode 字符，如需明确使用 Unicode 编码，请使用以下语法：

WITH 提供了编写辅助语句的方法，以便在更大的查询中使用。这些语句通常被称为公共表表达式（Common Table Expression，CTE），可以认为它定义了只存在于一个查询中的临时视图。

WITH 语法：

下面的示例定义了一个 CTE： orders_with_total ，并在 GROUP BY 查询中使用它。

SELECT 语句的一般语法为：

table_expression 可以是任何数据源（表、视图、VALUES 子句、多个表的 Join 结果、子查询）。下面的事例读取 Orders 表的所有列：

select_list 指定 * 表示解析所有的列，但是不建议在生产环境中使用，会降低性能，建议只查询需要的列：

查询可以使用 VALUES 子句，每个元组（Tuple）对应一个 Row，并且可以设置别名：

WHERE 语句可以过滤 Row：

可以对每行数据的指定列调用函数（内置、自定义函数，自定义函数必须提前注册）：

如果指定 SELECT DISTINCT，则将从结果集中删除重复行（每组重复中保留一行）。

对于流式查询，计算查询结果所需的状态（State）可能会无限增长。状态大小取决于不同行的数量。可<u>以为查询配置适当的状态生存时间（TTL），以防止状态大小过大。这可能会影响查询结果的正确性</u>。

Window 是流处理的核心。Windows 将流拆分为有限大小的片段应用计算。只有流处理支持。

Flink 1.13 提供了几个 Table-valued functions（TVF，区别于 Group Window Function），将表中的元素划分为 windows，包括：

- 滚动窗口（Tumbling windows）

- 滑动窗口（Hop, Sliding windows）

- 累加窗口（Cumulate windows）

- 会话窗口（Session windows，TVF 暂不支持）

每个元素在逻辑上可以属于多个窗口，具体取决于所使用的窗口函数。TVF 必须和聚合操作一起使用：

假设存在一个 Bid 表

指定一个固定大小的窗口，并且不重叠，语法：

设定一个10分钟大小的滚动窗口，

指定一个固定大小的窗口，设定滑动间隔，元素会被指定给多个窗口，语法：

设定一个10分钟大小，每5分钟滑动的窗口，

指定一个窗口的最大规模，按照指定时间间隔增长累加，直到达到窗口的最大规模，每次窗口增长会进行一次计算，可以理解为多次计算的滚动窗口，语法：

设定一个10分钟大小，每2分钟累计一次的窗口，

Flink Table API & SQL 基本操作

Flink 版本：1.13.5

本文主要展示了 Table API 和 SQL 程序的常见结构，如何创建注册 Table，查询 Table，以及如何输出 Table。

1. Table API & SQL 程序结构

在 Flink 中，Table API 和 SQL 可以看作联结在一起的一套 API，这套 API 的核心概念是一个可以用作 Query 输入和输出的表 Table。在我们程序中，输入数据可以定义成一张表，然后对这张表进行查询得到一张新的表，最后还可以定义一张用于输出的表，负责将处理结果写入到外部系统。

我们可以看到，程序的整体处理流程与 DataStream API 非常相似，也可以分为读取数据源(Source)、转换(Transform)、输出数据(Sink)三部分。只不过这里的输入输出操作不需要额外定义，只需要将用于输入和输出的表 Table 定义出来，然后进行转换查询就可以了。

SQL 程序基本架构如下：

// 创建执行环境 TableEnvironment
EnvironmentSettings settings = EnvironmentSettings
        .newInstance