当前位置：首页 > news >正文

flink-sql所有语法详解

news 来源：原创 2024/5/19 16:30:37

1. 版本说明

本文档介绍的各种flink sql的语法基于flink-1.13.x，flink版本低于1.13.x的用户，在sql运行出错误时，需要自行去flink官网查看对应版本的语法支持。

另外，flink新版本支持的语法，文档中会进行特殊标注，说明对应语法在 flink 哪个版本开始支持，但凡是没有特殊标注的，均支持flink-1.13.x及以上版本。

2. 其他

2.1. 技术

注意，在 flink sql 中，对表名、字段名、函数名等是严格区分大小写的，为了兼容 hive 等其他仓库，建议建表时，表名和字段名都采用下划线连接单词的方式，以避免大小写问题。

比如 hive ，是不区分大小写的，所有大写字母最终都会被系统转化为小写字母，此时使用 flink sql 去读写 hive ，出现大写字母时，会出现找不到表或字段的错误。

关键字是不区分大小写的，比如 insert、select、create等。

flink sql 中所有的字符串常量都需要使用英文单引号括起来，不要使用英文双引号以及中文符号。

3. SET

3.1. 介绍

SET语句用于修改配置或列出配置。

3.2. 语法

SET ('key' = 'value')

如果没有指定键和值，则只打印所有属性。否则，使用指定的键值对设置属性值。

3.3. 案例

Flink SQL> SET 'table.local-time-zone' = 'Europe/Berlin';
[INFO] Session property has been set.

Flink SQL> SET;
'table.local-time-zone' = 'Europe/Berlin'

4. RESET

4.1. 介绍

RESET语句用于将配置重置为默认值。

4.2. 语法

RESET ('key')

如果没有指定键，则将所有属性重置为默认值。否则，将指定的键重置为默认值。

4.3. 案例

Flink SQL> RESET 'table.planner';
[INFO] Session property has been reset.

Flink SQL> RESET;
[INFO] All session properties have been set to their default values.

5. CREATE

5.1. 介绍

CREATE 语句用于将 表/视图/函数 注册到当前或指定的 Catalog 中。已注册的表/视图/函数可以在SQL查询中使用。

Flink SQL目前支持以下CREATE语句：

CREATE TABLE
CREATE CATALOG
CREATE DATABASE
CREATE VIEW
CREATE FUNCTION

5.2. CREATE TABLE

语法概述：

CREATE [TEMPORARY] TABLE [IF NOT EXISTS] [catalog_name.][db_name.]table_name
(
{ <physical_column_definition> | <metadata_column_definition> | <computed_column_definition> }[ , ...n]
[ <watermark_definition> ]
[ <table_constraint> ][ , ...n]
)
[COMMENT table_comment]
[PARTITIONED BY (partition_column_name1, partition_column_name2, ...)]
WITH (key1=val1, key2=val2, ...)
[ LIKE source_table [( <like_options> )] ]

<physical_column_definition>:
column_name column_type [ <column_constraint> ] [COMMENT column_comment]

<column_constraint>:
[CONSTRAINT constraint_name] PRIMARY KEY NOT ENFORCED

<table_constraint>:
[CONSTRAINT constraint_name] PRIMARY KEY (column_name, ...) NOT ENFORCED

<metadata_column_definition>:
column_name column_type METADATA [ FROM metadata_key ] [ VIRTUAL ]

<computed_column_definition>:
column_name AS computed_column_expression [COMMENT column_comment]

<watermark_definition>:
WATERMARK FOR rowtime_column_name AS watermark_strategy_expression

<source_table>:
[catalog_name.][db_name.]table_name

<like_options>:
{
{ INCLUDING | EXCLUDING } { ALL | CONSTRAINTS | PARTITIONS }
| { INCLUDING | EXCLUDING | OVERWRITING } { GENERATED | OPTIONS | WATERMARKS }
}[, ...]

上面的语句创建了一个带有给定名称的表。如果catalog中已经存在同名的表，则会引发异常。

5.2.1. Columns（字段）

5.2.1.1. Physical / Regular Columns（物理/常规列）

物理列是数据库中已知的常规列。它们定义物理数据中字段的名称、类型和顺序。因此，物理列表示从外部系统读取和写入的有效负载。

连接器和格式转化使用这些列(按照定义的顺序)来配置自己。其他类型的列可以在物理列之间声明，但不会影响最终的物理模式。

下面的语句创建了一个只有常规列的表：

CREATE TABLE MyTable (
    `user_id` BIGINT,
    `name` STRING
) WITH (
...
);

5.2.1.2. Metadata Columns（元数据列）

元数据列是SQL标准的扩展，允许访问连接器和/或表中每一行的特定字段。元数据列由metadata关键字表示。例如，元数据列可以用来读取和写入Kafka记录的时间戳，以进行基于时间的操作。

连接器和格式文档列出了每个组件的可用元数据字段。在表的模式中声明元数据列是可选的。

下面的语句创建了一个表，其中包含引用元数据timestamp的附加元数据列：

CREATE TABLE MyTable (
    `user_id` BIGINT,
    `name` STRING,
    `record_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp' -- 读取和写入kafka记录的时间戳
) WITH (
    'connector' = 'kafka'
...
);

每个元数据字段都由基于字符串的键标识，并具有文档化的数据类型。例如，Kafka连接器暴露了一个元数据字段，该字段由键timestamp和数据类型TIMESTAMP_LTZ(3)标识，可以用于读写记录。

在上面的例子中，元数据列record_time成为表模式的一部分，可以像普通列一样进行转换和存储：

INSERT INTO MyTable SELECT user_id, name, record_time + INTERVAL '1' SECOND FROM MyTable;

为了方便起见，如果将列名直接用于标识元数据，则可以省略FROM子句：

CREATE TABLE MyTable (
    `user_id` BIGINT,
    `name` STRING,
    `timestamp` TIMESTAMP_LTZ(3) METADATA -- 使用列名作为元数据键
) WITH (
    'connector' = 'kafka'
...
);

为方便起见，如果列的数据类型与元数据字段的数据类型不同，可以显式指示强制类型转换，不过要求这两种数据类型是兼容的。

CREATE TABLE MyTable (
    `user_id` BIGINT,
    `name` STRING,
    `timestamp` BIGINT METADATA -- 转化timestamp类型为BIGINT
) WITH (
'connector' = 'kafka'
...
);

默认情况下，planner计划器会假定元数据列可以同时用于读写。然而在许多情况下，外部系统提供的元数据字段用于只读比可写更多。因此，可以使用VIRTUAL关键字将元数据列排除在持久化之外。

CREATE TABLE MyTable (
    `timestamp` BIGINT METADATA, -- query-to-sink schema的一部分
    `offset` BIGINT METADATA VIRTUAL, -- 不是query-to-sink schema的一部分
    `user_id` BIGINT,
    `name` STRING,
) WITH (
    'connector' = 'kafka'
...

);

在上面的示例中，偏移量是一个只读元数据列，并从query-to-sink schema中排除。因此，source-to-query模式(用于SELECT)和query-to-sink(用于INSERT INTO)模式不同:

source-to-query schema:
MyTable(`timestamp` BIGINT, `offset` BIGINT, `user_id` BIGINT, `name` STRING)
query-to-sink schema:
MyTable(`timestamp` BIGINT, `user_id` BIGINT, `name` STRING)

5.2.1.3. Computed Columns（计算列）

计算列是使用语法column_name AS computed_column_expression生成的虚拟列。

计算列可以引用同一表中声明的其他列的表达式，可以访问物理列和元数据列。列本身并不物理地存储在表中，列的数据类型通过给定的表达式自动派生，不需要手动声明。

计划器会将计算列转换为常规投影。对于优化或水印策略下推，计算列的实际计算可能会跨算子进行，并执行多次，或者在给定查询不需要的情况下跳过。例如，计算列可以定义为：

CREATE TABLE MyTable (
    `user_id` BIGINT,
    `price` DOUBLE,
    `quantity` DOUBLE,
    `cost` AS price * quanitity, -- 执行表达式并接收查询结果
) WITH (
    'connector' = 'kafka'
    ...
);

表达式可以是列、常量或函数的任意组合。表达式不能包含子查询。

计算列通常在Flink中用于在CREATE TABLE语句中定义时间属性。

可以通过proc AS PROCTIME()使用系统的PROCTIME()函数轻松定义处理时间属性。
事件时间属性timestamp可以在水印声明之前进行预处理。例如，如果原始字段不是TIMESTAMP(3)类型或嵌套在JSON字符串中，则可以使用计算列。

与虚拟元数据列类似，计算列被排除在持久化之外。因此，计算列不能是INSERT INTO语句的目标列。因此，source-to-query模式(用于SELECT)和query-to-sink(用于INSERT - INTO)模式不同：

source-to-query schema:
MyTable(`user_id` BIGINT, `price` DOUBLE, `quantity` DOUBLE, `cost` DOUBLE)
query-to-sink schema:
MyTable(`user_id` BIGINT, `price` DOUBLE, `quantity` DOUBLE)

5.2.2. WATERMARK

WATERMARK子句用于定义表的事件时间属性，其形式为WATERMARK FOR rowtime_column_name AS watermark_strategy_expression。

rowtime_column_name定义一个列，该列被标记为表的事件时间属性。该列必须为TIMESTAMP(3)类型，并且是模式中的顶级列。它可以是一个计算列。
watermark_strategy_expression定义了水印生成策略。它允许任意非查询表达式(包括计算列)来计算水印。表达式返回类型必须为TIMESTAMP(3)，表示从Epoch开始的时间戳。

返回的水印只有在非空且其值大于先前发出的本地水印时才会发出(以保持升序水印的规定)。框架会对每条记录执行水印生成表达式。框架将周期性地发出生成的最大水印。

如果当前水印与前一个相同，或为空，或返回的水印值小于上次发出的水印值，则不会发出新的水印。水印通过pipeline.auto-watermark-interval配置的时间间隔发出。

如果水印间隔为0ms，弱生成的水印不为空且大于上次发出的水印，则每条记录都发出一次水印。

当使用事件时间语义时，表必须包含事件时间属性和水印策略。

Flink提供了几种常用的水印策略：

严格递增时间戳：WATERMARK FOR rowtime_column AS rowtime_column
发出到目前为止观察到的最大时间戳的水印。时间戳大于最大时间戳的行不属于延迟。
升序时间戳：WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL ‘0.001’ SECOND
发出到目前为止观察到的最大时间戳减去1的水印。时间戳大于或等于最大时间戳的行不属于延迟。
时间戳：WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL ‘string’ timeUnit
发出到目前为止观察到的最大时间戳减去指定延迟的水印，例如：WATERMARK FOR rowtime_column AS rowtime_column - INTERVAL ‘5’ SECOND是一个延迟5秒的水印策略。

CREATE TABLE Orders (
user BIGINT,
product STRING,
order_time TIMESTAMP(3),
WATERMARK FOR order_time AS order_time - INTERVAL ‘5’ SECOND
) WITH ( . . . );

5.2.3. PRIMARY KEY

主键约束是Flink用于优化的一个提示。它告诉flink，指定的表或视图的一列或一组列是唯一的，它们不包含null。主列中的任何一列都不能为空。主键唯一地标识表中的一行。

主键约束可以与列定义(列约束)一起声明，也可以作为单行声明(表约束)。只能使用这两种方式之一，如果同时定义多个主键约束，则会引发异常。

有效性检查

SQL标准指定约束可以是强制的，也可以是不强制的。这将控制是否对传入/传出数据执行约束检查。Flink不保存数据，因此我们希望支持的唯一模式是not forced模式。确保查询执行的主键唯一性由用户负责。

注意:在CREATE TABLE语句中，主键约束会改变列的可空性，也就是说，一个有主键约束的列是不能为NULL的。

5.2.4. PARTITIONED BY

根据指定的列对已创建的表进行分区。如果将该表用作filesystem sink，则为每个分区创建一个目录。

5.2.5. WITH选项

用于创建表source/sink的表属性，属性通常用于查找和创建底层连接器。

表达式key1=val1的键和值都应该是字符串字面值。有关不同连接器的所有受支持的表属性，请参阅连接器中的详细信息。

表名可以是三种格式:

catalog_name.db_name.table_name
db_name.table_name
table_name。

对于catalog_name.db_name.Table_name，表将被注册到catalog名为“catalog_name”，数据库名为“db_name；
对于db_name.Table_name，表将注册到当前表执行环境的catalog和数据库名为“db_name”；
对于table_name，表将注册到表执行环境的当前catalog和数据库中。

注意：用CREATE table语句注册的表既可以用作表source，也可以用作表sink，我们不能决定它是用作源还是用作接收器，直到它在dml语句中被引用。

5.2.6. LIKE

LIKE子句是SQL特性的变体/组合。子句可用于基于现有表的定义创建表。此外，用户可以扩展原始表或排除其中的某些部分。与SQL标准相反，子句必须在CREATE语句的顶层定义。这是因为子句适用于定义的多个部分，而不仅仅适用于模式部分。

您可以使用该子句重用或覆盖某些连接器属性或向外部定义的表添加水印。例如，在Apache Hive中定义的表中添加水印。

下面为示例语句：

CREATE TABLE Orders (
  `user` BIGINT,
  product STRING,
  order_time TIMESTAMP(3)
) WITH (
  'connector' = 'kafka',
  'scan.startup.mode' = 'earliest-offset'
);

CREATE TABLE Orders_with_watermark (
  -- 增加水印定义
  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
  -- 覆盖startup-mode
  'scan.startup.mode' = 'latest-offset'
)
LIKE Orders;

生成的表Orders_with_watermark等价于用以下语句创建的表:

CREATE TABLE Orders_with_watermark (
  `user` BIGINT,
  product STRING,
  order_time TIMESTAMP(3),
  WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  'scan.startup.mode' = 'latest-offset'
);

表特性的合并逻辑可以通过like选项进行控制。

可以控制合并的特性有：

CONSTRAINTS：约束，比如主键和唯一键
GENERATED：计算列
METADATA：元数据列
OPTIONS：描述连接器和格式属性的连接器选项
PARTITIONS：表的分区
WATERMARKS：水印声明

有三种不同的合并策略：

INCLUDING：包含源表的特性，有重复的表项时失败，例如，如果两个表中都存在一个具有相同键的选项。
EXCLUDING：不包含源表的给定特性。
OVERWRITING：包含源表的特性，用新表的属性覆盖源表的重复项，例如，如果两个表中都存在一个具有相同键的选项，则使用当前语句中的选项。

此外，如果没有定义特定的策略，可以使用INCLUDING/EXCLUDING ALL选项来指定使用什么策略，例如，如果你使用EXCLUDING ALL INCLUDING WATERMARKS，则表示只有源表中的水印会被包含。

例子：

-- 存储在filesystem中的source表
CREATE TABLE Orders_in_file (
  `user` BIGINT,
  product STRING,
  order_time_string STRING,
  order_time AS to_timestamp(order_time)
)
PARTITIONED BY (`user`)
WITH (
  'connector' = 'filesystem',
  'path' = '...'
);
-- 想存储在kafka中的对应的表
CREATE TABLE Orders_in_kafka (
-- 增加水印定义
WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
  'connector' = 'kafka',
  ...
)
LIKE Orders_in_file (
  -- 不包含任何东西，除了需要的水印计算列。
  -- 我们不需要分区和文件系统选项这些kafka不接受的特性。
  EXCLUDING ALL
  INCLUDING GENERATED
);

如果没有提供like选项，INCLUDING ALL OVERWRITING OPTIONS 将作为默认选项使用。无法控制物理列归并行为。这些物理列将被合并，就像使用了INCLUDING策略一样。

source_table可以是复合标识符。因此，它可以是来自不同catalog或数据库的表，例如：

my_catalog.my_db.MyTable，指定来自MyCatalog和数据库my_db的表MyTable；
my_db.MyTable指定来自当前catalog和数据库my_db的表MyTable。

5.2.7. 案例代码

-- 读取kafka
CREATE TABLE kafka_bscc_bsnet_sendmsg (
    `time` bigint COMMENT '',
    `endpoint` string NULL COMMENT '主机名',
    `P999_delay` int NULL COMMENT 'P999分位的时延',
    `msg_trace_id` string NULL COMMENT '下发消息的 id',
    `error_count` int NULL COMMENT '业务方错误机器数量'
) with (
    'connector' = 'kafka',
    'topic' = 'this-is-a-topic',
    'properties.bootstrap.servers' = 'broker1:9092,broker2:9092,broker3:9092',
    'properties.group.id' = 'flinkSql.for-test',
    'scan.startup.mode' = 'earliest-offset',
    -- 格式化配置
    'format' = 'json',
    'json.ignore-parse-errors' = 'true',
    'json.encode.decimal-as-plain-number' = 'true'
)
;

5.3. CREATE CATALOG

CREATE CATALOG catalog_name WITH (
    key1=val1, key2=val2, ...
)

5.3.1. 参数说明

参数

必选

默认值

类型

描述

type

是

无

string

Catalog 的类型。创建 HiveCatalog 时，该参数必须设置为’hive’。

hive-conf-dir

否

无

string

指向包含 hive-site.xml 目录的 URI。该 URI 必须是 Hadoop 文件系统所支持的类型。
如果指定一个相对 URI，即不包含 scheme，则默认为本地文件系统。如果该参数没有指定，我们会在 class path 下查找hive-site.xml。

default-database

否

default

string

当一个catalog被设为当前catalog时，所使用的默认当前database。

hive-version

否

无

string

HiveCatalog 能够自动检测使用的 Hive 版本。我们建议不要手动设置 Hive 版本，除非自动检测机制失败。

hadoop-conf-dir

否

无

string

Hadoop 配置文件目录的路径。目前仅支持本地文件系统路径。我们推荐使用HADOOP_CONF_DIR环境变量来指定 Hadoop 配置。
因此仅在环境变量不满足您的需求时再考虑使用该参数，例如当您希望为每个 HiveCatalog 单独设置 Hadoop 配置时。

使用给定的目录属性创建目录。如果已经存在同名的目录，则会引发异常。

WITH选择
用于指定与此目录相关的额外信息的目录属性。表达式key1=val1的键和值都应该是字符串字面值。

注意，key和value都应该使用英文单引号括起来。

5.3.2. 案例代码

create catalog hive with (
    'type' = 'hive',
    'hadoop-conf-dir' = '/path/to/dir',
    'hive-conf-dir' = '/path/to/dir'
)
;

如果用户使用的是内存类型的 catalog ，也就是说没有创建 hive catalog ，则默认的 catalog 名称为default_catalog，默认的 database 名称为default_database。

在建表时，如果没有单独指定表所属的 catalog 和 database ，则使用上述默认的 catalog 和 database。

建议在建表时，不要指定 catalog 和 database 名称，这样比较方便。

如果用户使用的是 hive 类型的 catalog，也就是用户创建了 hive catalog ，并且使用了创建的 hive catalog（use catalog hive;），
则默认的 catalog 名称为创建的 hive catalog 名称。

比如上面的案例代码，catalog 名称就是hive，默认的 database 名称为 default。之后新建的表（非临时表），运行时将会出现在 hive 元数据中。

之后通过 HUE 等连接 hive 的工具，就可以通过show catete table table_name语句查看 flink 建表的元信息。

线上最佳实践

建议建表时，不指定 catalog 和 database 名称，以减少后续查询 sql 的便捷性。
如果使用了 hive catalog，则建表时，建议创建临时表，这样做有两点好处
1. 避免任务重启时出现标已存在的错误，当然可以添加if not exist来避免。
2. sql 代码中有建表语句，方便随时查看表的字段等信息。
不过在使用了 hive catalog的情况下，建表时创建了非临时表，在开发其他任务时，就可以不用编写建表语句了。

5.4. CREATE DATABASE

CREATE DATABASE [IF NOT EXISTS] [catalog_name.]db_name
[COMMENT database_comment]
WITH (key1=val1, key2=val2, ...)

使用给定的数据库属性创建数据库。如果目录中已经存在同名的数据库，则会引发异常。

IF NOT EXISTS
如果数据库已经存在，则不会发生任何事情。

WITH OPTIONS
用于指定与此数据库相关的额外信息的数据库属性。表达式key1=val1的键和值都应该是字符串字面值。

5.5. CREATE VIEW

CREATE [TEMPORARY] VIEW [IF NOT EXISTS] [catalog_name.][db_name.]view_name
[( columnName [, columnName ]* )] [COMMENT view_comment]
AS query_expression

使用给定的查询表达式创建视图。如果 catalog 中已经存在同名的视图，则会抛出异常。

TEMPORARY
创建具有目录和数据库名称空间并覆盖视图的临时视图。

IF NOT EXISTS
如果视图已经存在，则不会发生任何事情。

创建视图，可以将负责的查询 sql 进行拆分，以获取更好的阅读体验。

5.6. CREATE FUNCTION

CREATE [TEMPORARY|TEMPORARY SYSTEM] FUNCTION [IF NOT EXISTS] [catalog_name.][db_name.]function_name AS identifier [LANGUAGE JAVA|SCALA|PYTHON]

创建一个函数，该函数具有带有标识符和可选语言标记的catalog和数据库名称空间。如果目录中已经存在同名的函数，则会引发异常。

如果语言标记是JAVA/SCALA，则标识符是UDF的完整类路径。关于Java/Scala UDF的实现，请参考用户自定义函数。

如果语言标记是PYTHON，则标识符是UDF的完全限定名，例如pyflink.table.tests.test_udf.add。

有关Python UDF的实现，请参阅官网，这里暂不列出。

TEMPORARY
创建具有catalog和数据库名称空间并覆盖编目函数的临时编目函数。

TEMPORARY SYSTEM
创建没有命名空间并覆盖内置函数的临时系统函数。

IF NOT EXISTS
如果函数已经存在，则什么也不会发生。

LANGUAGE JAVA|SCALA|PYTHON
用于指导Flink运行时如何执行该函数的语言标记。目前只支持JAVA、SCALA和PYTHON，函数默认语言为JAVA。

5.6.1. 案例代码

create temporary function fetch_millisecond as 'cn.com.log.function.udf.time.FetchMillisecond' language java;

6. INSERT

6.1. 介绍

NSERT语句用于向表中添加行数据。

6.2. 通过select查询Insert数据

select查询结果可以通过使用insert子句插入到表中。

6.2.1. 语法

INSERT { INTO | OVERWRITE } [catalog_name.][db_name.]table_name [PARTITION part_spec] [column_list] select_statement
part_spec:
    (part_col_name1=val1 [, part_col_name2=val2, ...])
column_list:
    (col_name1 [, column_name2, ...])

OVERWRITE
INSERT OVERWRITE将覆盖表或分区中的任何现有数据。否则（INTO），将追加新的数据。

PARTITION
PARTITION子句指定插入语句的静态分区列。

COLUMN LIST
现在有表T(a INT, b INT, c INT)， flink支持

INSERT INTO T(c, b) SELECT x, y FROM S

查询的数据列‘x’将被写入列‘c’，查询的数据列‘y’将被写入列‘b’，并且列‘a’被设置为NULL（需保证列‘z’是可以为空的）。

overwrite 和 partition 关键字经常用于写入 hive 。

6.2.2. 案例

-- 创建一个分区表
CREATE TABLE country_page_view (user STRING, cnt INT, date STRING, country STRING)
PARTITIONED BY (date, country)
WITH (...)

-- 向静态分区(date='2019-8-30', country='China')追加数据行
INSERT INTO country_page_view PARTITION (date='2019-8-30', country='China')
SELECT user, cnt FROM page_view_source;

-- 向分区(date, country)追加数据行，静态data分区值为“2019-8-30”，country为动态分区，该分区值通过每行对应字段值动态获取
INSERT INTO country_page_view PARTITION (date='2019-8-30')
SELECT user, cnt, country FROM page_view_source;

-- 向静态分区(date='2019-8-30', country='China')覆盖数据
INSERT OVERWRITE country_page_view PARTITION (date='2019-8-30', country='China')
SELECT user, cnt FROM page_view_source;

-- 向分区(date, country)覆盖数据行，静态data分区值为“2019-8-30”，country为动态分区，该分区值通过每行对应字段值动态获取
INSERT OVERWRITE country_page_view PARTITION (date='2019-8-30')
SELECT user, cnt, country FROM page_view_source;

-- 向静态分区(date='2019-8-30', country='China')追加数据行，cnt字段值被设置为NULL
INSERT INTO country_page_view PARTITION (date='2019-8-30', country='China') (user)
SELECT user FROM page_view_source;

6.3. Insert values into tables

可以使用INSERT…VALUES语句将数据直接从SQL插入到表中。

6.3.1. 语法

INSERT { INTO | OVERWRITE } [catalog_name.][db_name.]table_name VALUES values_row [, values_row ...]
values_row:
    : (val1 [, val2, ...])

OVERWRITE
INSERT OVERWRITE将覆盖表中任何现有数据。否则，将追加新的数据。

6.3.2. 案例

CREATE TABLE students (name STRING, age INT, gpa DECIMAL(3, 2)) WITH (...);
INSERT INTO students
VALUES ('fred flintstone', 35, 1.28), ('barney rubble', 32, 2.32);

6.4. 运行多个insert

运行多个 insert 任务，在 flink UI 界面中，会体现出多个运行图。当然，如果你的多个 insert 语句读取了同一张表，或者是写入了同一张表，flink 则会对其优化，最后生成一张运行图。

7. USE

7.1. 介绍

USE语句用于设置当前数据库或catalog，或更改模块的解析顺序和启用状态。

7.2. USE CATALOG

USE CATALOG catalog_name

设置当前catalog。所有未显式指定catalog的后续命令都将使用此catalog。如果提供的catalog不存在，则抛出异常。默认当前catalog为default_catalog。

在flink sql中创建了hive catalog，再用use语句使用了hive catalog之后，flink 就连接上了 hive 的元数据，
之后的create创建的非临时表，其元数据就会被保存到 hive 的元数据。

如果不想在每个flink sql任务中重复创建连接外部系统的虚拟表，就可以只在第一个flink sql任务中创建一次表，之后的flink sql任务就不再需要编写建表语句了，
只要有创建hive catalog和use catalog hive语句即可。

7.3. USE MODULES

USE MODULES module_name1[, module_name2, ...]

按照声明的顺序设置已启用的模块。所有后续命令将解析启用模块中的元数据(函数/用户定义类型/规则等)，并遵循声明顺序。

模块在加载时被默认使用。如果没有使用USE modules语句，加载的模块将被禁用。默认加载和启用的模块是core。如果使用了该语句启动模块，则不在该语句中的模块都将被禁用。

7.3.1. 案例

use hive, core;

表示后续使用到的函数/用户定义类型/规则等，会先按照 hive 来解析，如果 hive 解析不了的，再用 flink 来解析。

7.4. USE

USE [catalog_name.]database_name

设置当前数据库。所有未显式指定数据库的后续命令都将使用此数据库。如果提供的数据库不存在，则抛出异常。默认的当前数据库是default_database。

8. SELECT

8.1. 介绍

select语句主要是从表中查询数据，然后将数据插入到其他表中。直接在页面中查看select的结果，目前平台还不支持。

单个反斜杠就可以作为转义字符使用，在select查询中可以直接使用。

8.2. 无表查询示例

SELECT supplier_id, rating, COUNT(*) AS total
FROM
    (VALUES
        ('supplier1', 'product1', 4),
        ('supplier1', 'product2', 3),
        ('supplier2', 'product3', 3),
        ('supplier2', 'product4', 4)
    ) AS Products(supplier_id, product_id, rating)
GROUP BY supplier_id, rating
;

可以将该查询结果作为临时视图，也可以作为子表使用，在测试自定义函数中会非常有用。

比如将查询的结果直接插入表：

insert into mysql_table
SELECT supplier_id, rating, COUNT(*) AS total
FROM
    (VALUES
        ('supplier1', 'product1', 4),
        ('supplier1', 'product2', 3),
        ('supplier2', 'product3', 3),
        ('supplier2', 'product4', 4)
    ) AS Products(supplier_id, product_id, rating)
GROUP BY supplier_id, rating
;

8.3. SQL提示

在流批处理任务中均可使用。

SQL提示可以与select语句一起使用，以改变运行时的配置。

在使用sql提示之前，需要通过SET语句启用动态表选项，也就是设置table.dynamic-table-options.enabled为true。

注：在1.13.x中，该配置默认为false，从1.14.x开始，该配置默认为 true。

8.3.1. 动态表操作

动态表（可以认为是任何表，虚拟表或者是hive中的表都可以）允许使用SQL提示在select查询中动态指定或覆盖表的选项配置，并且这种指定只会在当前的select语句中起作用。

8.3.1.1. 语法

flink sql采用了oracle风格的sql提示语法，如下所示：

table_path /*+ OPTIONS(key=val [, key=val]*) */
key:
    stringLiteralval:
    stringLiteral

8.3.1.2. 案例

CREATE TABLE kafka_table1 (id BIGINT, name STRING, age INT) WITH (...);
CREATE TABLE kafka_table2 (id BIGINT, name STRING, age INT) WITH (...);
-- 在查询中覆盖表选项
select id, name from kafka_table1 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
-- 在join时覆盖表选项
select * from
    kafka_table1 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ t1
    join
    kafka_table2 /*+ OPTIONS('scan.startup.mode'='earliest-offset') */ t2
    on t1.id = t2.id;
-- 覆盖insert的目标表选项
insert into kafka_table1 /*+ OPTIONS('sink.partitioner'='round-robin') */ 
select * from kafka_table2;
-- 通过 sql 提示指定的选项，如果和建表时通过 with 指定的选项重复的，sql 提示指定的选项会将其覆盖

8.4. WITH子句

在流批处理任务中均可使用。

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

语法：

WITH <with_item_definition> [ , ... ]
SELECT ... FROM ...;

<with_item_defintion>:
    with_item_name (column_name[, ...n]) AS ( <select_query> )

使用案例：

WITH orders_with_total AS (
    SELECT order_id, price + tax AS total
    FROM Orders
)
SELECT order_id, SUM(total)
FROM orders_with_total
GROUP BY order_id;

上面的with子句定义了orders_with_total，并且在group by子句中使用了它。

8.5. SELECT和WHERE

在流批模式任务中均可使用。

SELECT语句的一般语法为：

SELECT select_list FROM table_expression [ WHERE boolean_expression ];

table_expression可以引用任何数据源。它可以是一个现有表、视图或VALUES子句、多个现有表的连接结果或一个子查询。假设该表在catalog中可用，下面的语句将从Orders中读取所有行。

SELECT * FROM Orders;

select_list中的*号表示查询将会解析所有列。但是，在生产中不鼓励使用。相反，select_list可以指定手动可用列，或者使用可用列进行计算。入Orders表有名为order_id、price和tax的列，则可以编写以下查询:

SELECT order_id, price + tax FROM Orders

查询也可以通过VALUES子句使用内联数据。每个元组对应一行，可以提供一个别名来为每个列分配名称：

SELECT order_id, price FROM (VALUES (1, 2.0), (2, 3.1))  AS t (order_id, price);

可以根据WHERE子句筛选数据：

SELECT price + tax FROM Orders WHERE id = 10;

此外，可以在单行的列上调用内置和用户自定义的标量函数。用户自定义函数在使用前必须在目录中注册：

SELECT PRETTY_PRINT(order_id) FROM Orders;

8.6. SELECT DISTINCT

在流批模式任务中均可使用。

如果指定了SELECT DISTINCT，则会从结果集中删除所有重复的行(每组重复的行保留一行)：

SELECT DISTINCT id FROM Orders;

对于流式查询，计算查询结果所需的状态可能会无限增长。状态大小取决于不同的数据行数量。可以提供具有适当状态生存时间(TTL)的查询配置，以防止状态存储过大。
注意，这可能会影响查询结果的正确性。详细信息请参见查询配置。

8.7. 窗口表值函数TVF

只支持流式任务。

Windows是处理无限流的核心，Windows将流分成有限大小的桶，我们可以在桶上面进行计算。

Apache Flink提供了几个窗口表值函数(TVF)来将表中的元素划分为到窗口中以供用户进行处理，包括:

Tumble windows （滚动窗口）
Hop windows （滑动窗口）
Cumulate windows （累计窗口）
Session windows（会话窗口）(即将支持，目前1.13、1.14、1.15均未支持)

8.7.1. 窗口函数

Apache Flink提供了3个内置的窗口TVFs：TUMBLE、HOP和CUMULATE。窗口TVF的返回值是一个新的关系，它包括原来关系的所有列，以及另外3列，名为window_start，window_end
，window_time来表示分配的窗口。

window_time字段是窗口TVF执行之后的一个时间属性，可以在后续基于时间的操作中使用。window_time的值总是等于window_end - 1ms。

8.7.1.1. TUMBLE

滚动窗口函数将每个元素分配给指定大小的窗口,滚动窗口的大小是固定的，并且不会重叠。假设指定了一个大小为5分钟的滚动窗口，在这种情况下，Flink将计算当前窗口，并每5分钟启动一个新窗口，如下图所示。
在这里插入图片描述
TUMBLE函数根据时间属性列为表的每一行分配一个窗口。TUMBLE的返回值是一个新的关系，它包括原来表的所有列以及另外3列“window_start”，“window_end”，“window_time”来表示分配的窗口。

原表中的原始时间字段将是窗口TVF函数之后的常规时间列。TUMBLE函数需要三个参数:

-- 1.13.x
TUMBLE(TABLE data, DESCRIPTOR(timecol), size)
-- 从1.14.x开始，增加了一个可选的参数
TUMBLE(TABLE data, DESCRIPTOR(timecol), size [, offset ])

data：表名，该表必须有一列类型为时间戳，也就是TIMESTAMP类型。
timecol：列名，表示该列数据映射到滚动窗口。
size：指定滚动窗口的窗口大小。
offset：从1.14.x开始支持，可选参数，用于指定窗口开始移动的offset，也就是指定产生窗口的时间点。
比如窗口时间为5分钟，指定开始移动的 offset 为1分钟，则触发的窗口如下：[1分钟, 6分钟)、[6分钟, 11分钟)、…。

下面是一个对Bid表的调用示例：

-- 表必须有时间字段，比如下表中的 `bidtime` 字段。
Flink SQL> desc Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

Flink SQL> SELECT * FROM Bid;
+------------------+-------+------+
|          bidtime | price | item |
+------------------+-------+------+
| 2020-04-15 08:05 |  4.00 | C    |
| 2020-04-15 08:07 |  2.00 | A    |
| 2020-04-15 08:09 |  5.00 | D    |
| 2020-04-15 08:11 |  3.00 | B    |
| 2020-04-15 08:13 |  1.00 | E    |
| 2020-04-15 08:17 |  6.00 | F    |
+------------------+-------+------+

-- 注意：目前flink不支持单独使用表值窗口函数，表值窗口函数应该和聚合操作一起使用，这个示例只是展示语法以及通过表值函数产生数据
SELECT * 
FROM 
    TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
    );
-- 或者是和参数名称一起使用。注意：DATA参数必须是第一个
SELECT * 
FROM 
    TABLE(
        TUMBLE(
            DATA => TABLE Bid,
            TIMECOL => DESCRIPTOR(bidtime),
            SIZE => INTERVAL '10' MINUTES
        )
    );
+------------------+-------+------+------------------+------------------+-------------------------+
|          bidtime | price | item |     window_start |       window_end |            window_time  |
+------------------+-------+------+------------------+------------------+-------------------------+
| 2020-04-15 08:05 |  4.00 | C    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:07 |  2.00 | A    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:09 |  5.00 | D    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
| 2020-04-15 08:13 |  1.00 | E    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
| 2020-04-15 08:17 |  6.00 | F    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
+------------------+-------+------+------------------+------------------+-------------------------+

从上面的结果可以看到，原始表的6行数据被分配到3个窗口中，每个滚动窗口是时间间隔为10分钟，窗口时间window_time为对应窗口结束时间-1ms。

-- 在滚动窗口表上执行聚合函数
SELECT window_start, window_end, SUM(price)
FROM 
    TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end;
+------------------+------------------+-------+
|     window_start |       window_end | price |
+------------------+------------------+-------+
| 2020-04-15 08:00 | 2020-04-15 08:10 | 11.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | 10.00 |
+------------------+------------------+-------+

注意:为了更好地理解窗口的行为，我们简化了时间戳值的显示，不显示秒后面的零。
如果时间类型是timestamp(3)，在Flink SQL Client中，2020-04-15 08:05应该显示为2020-04-15 08:05:00.000。

8.7.1.2. HOP

HOP函数将元素分配给固定长度的窗口。和TUMBLE窗口功能一样，窗口的大小由窗口大小参数来配置，另一个窗口滑动参数控制跳跃窗口启动的频率，类似于 stream api 中的滑动窗口。

因此，如果滑动小于窗口大小，跳跃窗口就会重叠。在本例中，元素被分配给多个窗口。跳跃窗口也被称为“滑动窗口”。

例如，10分钟大小的窗口，滑动5分钟。这样，每5分钟就会得到一个窗口，窗口包含在最近10分钟内到达的事件，如下图所示。
在这里插入图片描述
HOP函数窗口会覆盖指定大小区间内的数据行，并根据时间属性列移动。

HOP的返回值是一个新的关系，它包括原来关系的所有列，以及“window_start”、“window_end”、“window_time”来表示指定的窗口。原表的原始的时间属性列“timecol”将是执行TVF后的常规时间戳列。

HOP接受四个必需的参数：

-- 1.13.x
HOP(TABLE data, DESCRIPTOR(timecol), slide, size)
-- 从1.14.x开始，增加了一个可选的参数
HOP(TABLE data, DESCRIPTOR(timecol), slide, size [, offset ])

data：表名，该表必须有一列类型为时间戳，也就是TIMESTAMP类型。
timecol：列名，表示该列数据映射到滑动窗口。
slide：滑动时间，指定连续滑动窗口之间的间隔时间。
size：指定滑动窗口的窗口大小。
offset：从1.14.x开始支持，可选参数，用于指定窗口开始移动的offset，也就是指定产生窗口的时间点。
比如窗口时间为5分钟，指定开始移动的 offset 为1分钟，则触发的窗口如下：[1分钟, 6分钟)、[6分钟, 11分钟)、…。

下面是一个对Bid表的调用示例：

-- 注意：目前flink不支持单独使用表值窗口函数，表值窗口函数应该和聚合操作一起使用，这个示例只是展示语法以及通过表值函数产生数据
SELECT * 
    FROM TABLE(
        HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES)
    );
-- 或者是和参数名称一起使用。注意：DATA参数必须是第一个
SELECT * 
FROM TABLE(
    HOP(
        DATA => TABLE Bid,
        TIMECOL => DESCRIPTOR(bidtime),
        SLIDE => INTERVAL '5' MINUTES,
        SIZE => INTERVAL '10' MINUTES
    )
);
+------------------+-------+------+------------------+------------------+-------------------------+
|          bidtime | price | item |     window_start |       window_end |           window_time   |
+------------------+-------+------+------------------+------------------+-------------------------+
| 2020-04-15 08:05 |  4.00 | C    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:05 |  4.00 | C    | 2020-04-15 08:05 | 2020-04-15 08:15 | 2020-04-15 08:14:59.999 |
| 2020-04-15 08:07 |  2.00 | A    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:07 |  2.00 | A    | 2020-04-15 08:05 | 2020-04-15 08:15 | 2020-04-15 08:14:59.999 |
| 2020-04-15 08:09 |  5.00 | D    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:09 |  5.00 | D    | 2020-04-15 08:05 | 2020-04-15 08:15 | 2020-04-15 08:14:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:05 | 2020-04-15 08:15 | 2020-04-15 08:14:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
| 2020-04-15 08:13 |  1.00 | E    | 2020-04-15 08:05 | 2020-04-15 08:15 | 2020-04-15 08:14:59.999 |
| 2020-04-15 08:13 |  1.00 | E    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
| 2020-04-15 08:17 |  6.00 | F    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
| 2020-04-15 08:17 |  6.00 | F    | 2020-04-15 08:15 | 2020-04-15 08:25 | 2020-04-15 08:24:59.999 |
+------------------+-------+------+------------------+------------------+-------------------------+

从上面的结果可以看出，由于窗口有重叠，所有很多数据都属于两个窗口。

-- 在滑动窗口表上运行聚合函数
SELECT window_start, window_end, SUM(price)
FROM 
    TABLE(
        HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end;
+------------------+------------------+-------+
|     window_start |       window_end | price |
+------------------+------------------+-------+
| 2020-04-15 08:00 | 2020-04-15 08:10 | 11.00 |
| 2020-04-15 08:05 | 2020-04-15 08:15 | 15.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | 10.00 |
| 2020-04-15 08:15 | 2020-04-15 08:25 |  6.00 |
+------------------+------------------+-------+

8.7.1.3. CUMULATE

累积窗口，或者叫做渐进式窗口，在某些情况下是非常有用的，例如在固定的窗口间隔内早期触发滚动窗口。

例如，仪表板显示当天的实时总UV数，需要从每天的00:00开始到累计每分钟的UV值，10:00的UV值表示00:00到10:00的UV总数，这就可以通过累积窗口轻松有效地实现。

CUMULATE函数将元素分配给窗口，这些窗口在初始步长间隔内覆盖行数据，并且每一步都会扩展到一个更多的步长(保持窗口开始时间为固定值)，直到最大窗口大小。

可以把CUMULATE函数看作是先应用具有最大窗口大小的TUMBLE窗口，然后把每个滚动窗口分成几个窗口，每个窗口的开始和结束都有相同的步长差。所以累积窗口有重叠，而且没有固定的大小。

例如有一个累积窗口，1小时的步长和1天的最大大小，将获得窗口:[00:00,01:00)，[00:00,02:00)，[00:00,03:00)，…，[00:00,24:00)，每天都如此。
在这里插入图片描述
累积窗口基于时间属性列分配窗口。CUMULATE的返回值是一个新的关系，它包括原来关系的所有列，另外还有3列，分别是“window_start”、“window_end”、“window_time”，表示指定的窗口。

原始的时间属性“timecol”将是窗口TVF之后的常规时间戳列。

CUMULATE接受四个必需的参数。

-- 1.13.x
CUMULATE(TABLE data, DESCRIPTOR(timecol), step, size)
-- 从1.14.x开始，增加了一个可选的参数
CUMULATE(TABLE data, DESCRIPTOR(timecol), step, size [, offset ])

data：表名，该表必须有一列类型为时间戳，也就是TIMESTAMP类型。
timecol：列名，表示该列数据映射到累计窗口。
step：步长，指定连续累积窗口结束时间之间增加的窗口大小的时间间隔。
size：指定累积窗口的窗口大小。大小必须是步长的整数倍。
offset：从1.14.x开始支持，可选参数，用于指定窗口开始移动的offset，也就是指定产生窗口的时间点。
比如窗口时间为5分钟，指定开始移动的 offset 为1分钟，则触发的窗口如下：[1分钟, 6分钟)、[6分钟, 11分钟)、…。

下面是一个对Bid表的调用示例：

-- 注意：目前flink不支持单独使用表值窗口函数，表值窗口函数应该和聚合操作一起使用，这个示例只是展示语法以及通过表值函数产生数据
SELECT * 
FROM 
    TABLE(
        CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '2' MINUTES, INTERVAL '10' MINUTES)
    );
-- 或者是和参数名称一起使用。注意：DATA参数必须是第一个
SELECT * 
FROM 
    TABLE(
        CUMULATE(
            DATA => TABLE Bid,
            TIMECOL => DESCRIPTOR(bidtime),
            STEP => INTERVAL '2' MINUTES,
            SIZE => INTERVAL '10' MINUTES
        )
    );
+------------------+-------+------+------------------+------------------+-------------------------+
|          bidtime | price | item |     window_start |       window_end |            window_time  |
+------------------+-------+------+------------------+------------------+-------------------------+
| 2020-04-15 08:05 |  4.00 | C    | 2020-04-15 08:00 | 2020-04-15 08:06 | 2020-04-15 08:05:59.999 |
| 2020-04-15 08:05 |  4.00 | C    | 2020-04-15 08:00 | 2020-04-15 08:08 | 2020-04-15 08:07:59.999 |
| 2020-04-15 08:05 |  4.00 | C    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:07 |  2.00 | A    | 2020-04-15 08:00 | 2020-04-15 08:08 | 2020-04-15 08:07:59.999 |
| 2020-04-15 08:07 |  2.00 | A    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:09 |  5.00 | D    | 2020-04-15 08:00 | 2020-04-15 08:10 | 2020-04-15 08:09:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:10 | 2020-04-15 08:12 | 2020-04-15 08:11:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:10 | 2020-04-15 08:14 | 2020-04-15 08:13:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:10 | 2020-04-15 08:16 | 2020-04-15 08:15:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:10 | 2020-04-15 08:18 | 2020-04-15 08:17:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
| 2020-04-15 08:13 |  1.00 | E    | 2020-04-15 08:10 | 2020-04-15 08:14 | 2020-04-15 08:13:59.999 |
| 2020-04-15 08:13 |  1.00 | E    | 2020-04-15 08:10 | 2020-04-15 08:16 | 2020-04-15 08:15:59.999 |
| 2020-04-15 08:13 |  1.00 | E    | 2020-04-15 08:10 | 2020-04-15 08:18 | 2020-04-15 08:17:59.999 |
| 2020-04-15 08:13 |  1.00 | E    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
| 2020-04-15 08:17 |  6.00 | F    | 2020-04-15 08:10 | 2020-04-15 08:18 | 2020-04-15 08:17:59.999 |
| 2020-04-15 08:17 |  6.00 | F    | 2020-04-15 08:10 | 2020-04-15 08:20 | 2020-04-15 08:19:59.999 |
+------------------+-------+------+------------------+------------------+-------------------------+

-- 在窗口表上运行聚合函数
SELECT window_start, window_end, SUM(price)
FROM
    TABLE(
        CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '2' MINUTES, INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end;
+------------------+------------------+-------+
|     window_start |       window_end | price |
+------------------+------------------+-------+
| 2020-04-15 08:00 | 2020-04-15 08:06 |  4.00 |
| 2020-04-15 08:00 | 2020-04-15 08:08 |  6.00 |
| 2020-04-15 08:00 | 2020-04-15 08:10 | 11.00 |
| 2020-04-15 08:10 | 2020-04-15 08:12 |  3.00 |
| 2020-04-15 08:10 | 2020-04-15 08:14 |  4.00 |
| 2020-04-15 08:10 | 2020-04-15 08:16 |  4.00 |
| 2020-04-15 08:10 | 2020-04-15 08:18 | 10.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | 10.00 |
+------------------+------------------+-------+

8.7.2. Window Offset

从1.14.x开始支持。

offset 是一个可选参数，该参数可以被用于更改窗口的分配，该参数的值应该是一个正数或者是负数。

默认的 window offset 为0。如果设置不同的 offset 值，则同一条数据可能会被分配到不同的窗口。

下面列出的是时间戳为 2021-06-30 00:00:04 的数据将被分配到哪个以10分钟为窗口大小的窗口：

如果 offset 值为 -16 MINUTE，数据将会被分配给 [2021-06-29 23:54:00, 2021-06-30 00:04:00) 窗口。
如果 offset 值为 -6 MINUTE，数据将会被分配给 [2021-06-29 23:54:00, 2021-06-30 00:04:00) 窗口。
如果 offset 值为 -4 MINUTE，数据将会被分配给 [2021-06-29 23:56:00, 2021-06-30 00:06:00) 窗口。
如果 offset 值为 0，数据将会被分配给 [2021-06-30 00:00:00, 2021-06-30 00:10:00) 窗口。
如果 offset 值为 4 MINUTE，数据将会被分配给 [2021-06-29 23:54:00, 2021-06-30 00:04:00) 窗口。
如果 offset 值为 6 MINUTE，数据将会被分配给 [2021-06-29 23:56:00, 2021-06-30 00:06:00) 窗口。
如果 offset 值为 16 MINUTE，数据将会被分配给 [2021-06-29 23:56:00, 2021-06-30 00:06:00) 窗口。

通过上述案例可以看到，有些窗口 offset 参数值可能会得到相同的窗口分配。

比如： -16 MINUTE, -6 MINUTE 和4 MINUTE ，对于 10 MINUTE 大小的窗口，会得到相同滚动窗口。

注意：窗口的 offset 只用于改变窗口的分配，而不会影响水印。

下面是一个在滚动窗口中使用 offset 的 sql 案例：

-- 注意：目前 flink 不支持直接使用窗口表值函数。
-- 窗口表值函数应该被用于聚合操作，下面的例子只是用于展示语法以及表值函数产生的数据。
Flink SQL> SELECT *
FROM
    TABLE(
    TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES, INTERVAL '1' MINUTES)
    );
-- 或者指定参数名称
-- 注意：DATA 参数必须放在最开始
Flink SQL> SELECT *
FROM 
    TABLE(
        TUMBLE(
            DATA => TABLE Bid,
            TIMECOL => DESCRIPTOR(bidtime),
            SIZE => INTERVAL '10' MINUTES,
            OFFSET => INTERVAL '1' MINUTES
        )
    );
+------------------+-------+------+------------------+------------------+-------------------------+
|          bidtime | price | item |     window_start |       window_end |            window_time  |
+------------------+-------+------+------------------+------------------+-------------------------+
| 2020-04-15 08:05 |  4.00 | C    | 2020-04-15 08:01 | 2020-04-15 08:11 | 2020-04-15 08:10:59.999 |
| 2020-04-15 08:07 |  2.00 | A    | 2020-04-15 08:01 | 2020-04-15 08:11 | 2020-04-15 08:10:59.999 |
| 2020-04-15 08:09 |  5.00 | D    | 2020-04-15 08:01 | 2020-04-15 08:11 | 2020-04-15 08:10:59.999 |
| 2020-04-15 08:11 |  3.00 | B    | 2020-04-15 08:11 | 2020-04-15 08:21 | 2020-04-15 08:20:59.999 |
| 2020-04-15 08:13 |  1.00 | E    | 2020-04-15 08:11 | 2020-04-15 08:21 | 2020-04-15 08:20:59.999 |
| 2020-04-15 08:17 |  6.00 | F    | 2020-04-15 08:11 | 2020-04-15 08:21 | 2020-04-15 08:20:59.999 |
+------------------+-------+------+------------------+------------------+-------------------------+

-- 在滚动窗口结果表上使用聚合函数
Flink SQL> SELECT window_start, window_end, SUM(price)
FROM
    TABLE(
            TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES, INTERVAL '1' MINUTES)
    )
GROUP BY window_start, window_end;
+------------------+------------------+-------+
|     window_start |       window_end | price |
+------------------+------------------+-------+
| 2020-04-15 08:01 | 2020-04-15 08:11 | 11.00 |
| 2020-04-15 08:11 | 2020-04-15 08:21 | 10.00 |
+------------------+------------------+-------+

注意：为了更好的理解窗口的行为，我们简单的展示了 timestamp 类型的值，并没有展示后面的0。
比如 2020-04-15 08:05 ，如果类型为 TIMESTAMP(3)，准确来说，在 sql client 中应该被展示为 2020-04-15 08:05:00.000。

8.8. 窗口聚合

8.8.1. 窗口表值函数TVF聚合

只支持流式任务。

在group by子句中定义的窗口聚合函数可以使用通过窗口表值聚合函数的结果表中的“window_start”和“window_end”列。
就像使用常规GROUP BY子句的查询一样，使用GROUP BY窗口聚合的查询会给每个组计算出单个结果行。

SELECT ...
FROM <windowed_table> -- 接受通过窗口表值函数TVF生成的表
GROUP BY window_start, window_end, ...

与连续流表上的其他聚合不同，窗口聚合不发出中间结果，而只发出最终结果，即窗口结束之后的总聚合。此外，当不再需要时，窗口聚合会清除所有中间状态。

8.8.1.1. 窗口表值函数TVF

Flink支持TUMBLE、HOP和CUMULATE类型的窗口聚合，它们可以定义在事件时间或处理时间属性上。

下面是一些TUMBLE、HOP和CUMULATE窗口聚合的例子。

-- 表必须有时间属性列，比如下面表中的`bidtime`列。
Flink SQL> desc Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
| supplier_id |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

Flink SQL> SELECT * FROM Bid;
+------------------+-------+------+-------------+
|          bidtime | price | item | supplier_id |
+------------------+-------+------+-------------+
| 2020-04-15 08:05 | 4.00  | C    | supplier1   |
| 2020-04-15 08:07 | 2.00  | A    | supplier1   |
| 2020-04-15 08:09 | 5.00  | D    | supplier2   |
| 2020-04-15 08:11 | 3.00  | B    | supplier2   |
| 2020-04-15 08:13 | 1.00  | E    | supplier1   |
| 2020-04-15 08:17 | 6.00  | F    | supplier2   |
+------------------+-------+------+-------------+

-- 滚动窗口聚合
SELECT window_start, window_end, SUM(price)
FROM 
    TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end;
+------------------+------------------+-------+
|     window_start |       window_end | price |
+------------------+------------------+-------+
| 2020-04-15 08:00 | 2020-04-15 08:10 | 11.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | 10.00 |
+------------------+------------------+-------+

-- 滑动窗口聚合
SELECT window_start, window_end, SUM(price)
FROM 
    TABLE(
        HOP(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES, INTERVAL '10' MINUTES))
GROUP BY window_start, window_end;
+------------------+------------------+-------+
|     window_start |       window_end | price |
+------------------+------------------+-------+
| 2020-04-15 08:00 | 2020-04-15 08:10 | 11.00 |
| 2020-04-15 08:05 | 2020-04-15 08:15 | 15.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | 10.00 |
| 2020-04-15 08:15 | 2020-04-15 08:25 | 6.00  |
+------------------+------------------+-------+

-- 累计窗口聚合
SELECT window_start, window_end, SUM(price)
FROM 
    TABLE(
        CUMULATE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '2' MINUTES, INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end;
+------------------+------------------+-------+
|     window_start |       window_end | price |
+------------------+------------------+-------+
| 2020-04-15 08:00 | 2020-04-15 08:06 | 4.00  |
| 2020-04-15 08:00 | 2020-04-15 08:08 | 6.00  |
| 2020-04-15 08:00 | 2020-04-15 08:10 | 11.00 |
| 2020-04-15 08:10 | 2020-04-15 08:12 | 3.00  |
| 2020-04-15 08:10 | 2020-04-15 08:14 | 4.00  |
| 2020-04-15 08:10 | 2020-04-15 08:16 | 4.00  |
| 2020-04-15 08:10 | 2020-04-15 08:18 | 10.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | 10.00 |
+------------------+------------------+-------+

注意:为了更好地理解窗口的行为，我们简化了时间戳值的显示，以不显示秒小数点后面的零，
例如，如果类型是timestamp(3)，在Flink SQL Client中，2020-04-15 08:05应该显示为2020-04-15 08:05:00.000。

8.8.1.2. GROUPING SETS

窗口聚合也支持GROUPING SETS语法。GROUPING SETS允许进行比标准GROUP BY更复杂的分组操作。行按每个指定的分组集单独分组，并为每个分组计算聚合，就像简单的group by子句一样。

带有GROUPING SETS的窗口聚合要求window_start和window_end列必须在GROUP BY子句中，但不能在GROUPING SETS子句中。

SELECT window_start, window_end, supplier_id, SUM(price) as price
FROM 
    TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end, GROUPING SETS ((supplier_id), ());
+------------------+------------------+-------------+-------+
|     window_start |       window_end | supplier_id | price |
+------------------+------------------+-------------+-------+
| 2020-04-15 08:00 | 2020-04-15 08:10 |      (NULL) | 11.00 |
| 2020-04-15 08:00 | 2020-04-15 08:10 |   supplier2 |  5.00 |
| 2020-04-15 08:00 | 2020-04-15 08:10 |   supplier1 |  6.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 |      (NULL) | 10.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 |   supplier2 |  9.00 |
| 2020-04-15 08:10 | 2020-04-15 08:20 |   supplier1 |  1.00 |
+------------------+------------------+-------------+-------+

GROUPING SETS的每个子列表可以指定零个或多个列或表达式，并且解释方式与直接写在GROUP BY子句相同。空分组集意味着将所有行聚合为单个组，即使没有输入行，该组也会输出。

对于GROUPING SETS中的子集，如果没有指定任何数据列或表达式，将会使用NULL值来代替，表示对窗口时间内的全量数据进行聚合。

8.8.1.2.1. ROLLUP

ROLLUP是一种用于指定通用分组集类型的简写符号。它表示给定的表达式列表，前缀列表和空列表。

前缀列表：也就是说，子列表是指定的所有字段，然后每次去掉最后面一个字段而生成的表达式列表，示例如下：

rollup(s1, s2, s3)
s1, s2, s3
s1, s2, null
s1, null, null
null, null, null

带有ROLLUP的窗口聚合要求window_start和window_end列必须在GROUP BY子句中，而不是在ROLLUP子句中。

例如，下面的查询与上面的查询等价。

SELECT window_start, window_end, supplier_id, SUM(price) as price
FROM 
    TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end, ROLLUP (supplier_id);

8.8.1.2.2. CUBE

CUBE是一种用于指定公共分组集类型的简写符号。它表示给定的列表及其所有可能的子集，包括空列表。

使用CUBE的窗口聚合要求window_start和window_end列必须在GROUP BY子句中，而不是在CUBE子句中。

例如，下面两个查询是等价的。

SELECT window_start, window_end, item, supplier_id, SUM(price) as price
FROM 
    TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end, CUBE (supplier_id, item);
------------------------------------------------------------------------------
SELECT window_start, window_end, item, supplier_id, SUM(price) as price
FROM 
    TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end,
GROUPING SETS (
    (supplier_id, item),
    (supplier_id      ),
    (             item),
    (                 )
)

8.8.1.3. select分组窗口开始结束时间

可以使用分组的window_start和window_end列来作为组窗口的开始和结束时间戳。

8.8.1.4. 层叠窗口聚合

window_start和window_end列是常规的时间戳列，而不是时间属性。因此，它们不能在随后的基于时间的操作中用作时间属性。为了传播时间属性，需要在GROUP BY子句中添加window_time列。

window_time是窗口表值函数TVF产生的第三列，它是指定窗口的时间属性，比窗口结束时间早1毫秒。将window_time添加到GROUP BY子句中，使得window_time也成为可以选择的时间列。

然后，查询就可以将此列用于后续基于时间的操作，例如层叠窗口聚合和窗口TopN。

下面代码显示了层叠窗口聚合用法，其中第一个窗口聚合函数传播第二个窗口聚合的时间属性。

-- 对每个supplier_id进行5分钟的滚动窗口计算
CREATE VIEW window1 AS
SELECT window_start, window_end, window_time as rowtime, SUM(price) as partial_price
FROM 
    TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '5' MINUTES)
    )
GROUP BY supplier_id, window_start, window_end, window_time;

-- 在上面的窗口结果基础上进行10分钟的窗口滚动计算
SELECT window_start, window_end, SUM(partial_price) as total_price
FROM 
    TABLE(
        TUMBLE(TABLE window1, DESCRIPTOR(rowtime), INTERVAL '10' MINUTES)
    )
GROUP BY window_start, window_end;

8.9. 分组聚合

可用于流批任务。

像大多数数据系统一样，Apache Flink支持聚合函数：内置的和用户定义的。用户定义函数在使用前必须在catalog中注册。

聚合函数通过多个输入行计算单个结果。例如，在一组行数据上计算COUNT、SUM、AVG(平均)、MAX(最大)和MIN(最小)的聚合。

SELECT COUNT(*) FROM Orders;

flink的流查询是永远不会终止的连续查询。该查询会根据输入表的新数据来更新结果表。对于上面的查询，每次在Orders表中插入新行时，Flink都会输出一个更新的计数。

Apache Flink支持用于聚合数据的标准GROUP BY子句。

SELECT COUNT(*)
FROM Orders
GROUP BY order_id;

对于流式查询，计算查询结果所需的状态可能会无限增长。状态大小取决于组的数量以及聚合函数的数量和类型。可以配置查询的状态生存时间(TTL)，以防止状态大小过大。但这可能会影响查询结果的正确性。
详细信息请参见查询配置。

Apache Flink为Group Aggregation提供了一系列性能调优方法，请参阅更多的性能调优。

8.9.1. DISTINCT聚合

有些聚合需要在调用聚合函数之前删除重复值。下面的示例计算Orders表中不同order_ids的数量，而不是总行数。

SELECT COUNT(DISTINCT order_id) FROM Orders;

对于流式查询，计算查询结果所需的状态可能无限增长。状态大小主要取决于不同的行数和组维护的时间，短时间的窗口组聚合不是问题。可以配置查询的状态生存时间(TTL)，以防止状态大小过大。
注意，这可能会影响查询结果的正确性。详细信息请参见查询配置。

8.9.2. GROUPING SETS

grouping sets可以执行比标准GROUP BY更复杂的分组操作。行数据按每个分组集单独分组，并为每个分组计算聚合函数，就像简单的group by子句一样。

SELECT supplier_id, rating, COUNT(*) AS total
FROM 
    (VALUES
        ('supplier1', 'product1', 4),
        ('supplier1', 'product2', 3),
        ('supplier2', 'product3', 3),
        ('supplier2', 'product4', 4)
    )
AS Products(supplier_id, product_id, rating)
GROUP BY GROUPING SETS ((supplier_id, rating), (supplier_id), ());
+-------------+--------+-------+
| supplier_id | rating | total |
+-------------+--------+-------+
|   supplier1 |      4 |     1 |
|   supplier1 | (NULL) |     2 |
|      (NULL) | (NULL) |     4 |
|   supplier1 |      3 |     1 |
|   supplier2 |      3 |     1 |
|   supplier2 | (NULL) |     2 |
|   supplier2 |      4 |     1 |
+-------------+--------+-------+

GROUPING SETS的每个子列表可以指定零个或多个列或表达式，并且其解释方式与直接在GROUP BY子句中使用相同。空分组集意味着将所有行聚合为单个组，即使没有输入行，该组也会输出。

对于分组中集中未出现的列或表达式，会使用NULL进行替换，如上图所示。

对于流式查询，计算查询结果所需的状态可能无限增长。状态大小取决于组集的数量和聚合函数的类型。可以配置查询的状态生存时间(TTL)，以防止状态大小过大。注意，这可能会影响查询结果的正确性。
详细信息请参见查询配置。

8.9.2.1. ROLLUP

ROLLUP是一种用于指定通用分组集类型的简单用法。它表示给定的表达式列表、前缀列表、空列表。例如，下面的查询与上面的查询等价。

SELECT supplier_id, rating, COUNT(*)
FROM 
    (VALUES
        ('supplier1', 'product1', 4),
        ('supplier1', 'product2', 3),
        ('supplier2', 'product3', 3),
        ('supplier2', 'product4', 4)
    )
AS Products(supplier_id, product_id, rating)
GROUP BY ROLLUP (supplier_id, rating);

8.9.2.2. CUBE

CUBE是一种用于指定公共分组集类型的简单用法。它表示给定的列表及其所有可能的子集。例如，下面两个查询是等价的。

SELECT supplier_id, rating, product_id, COUNT(*)
FROM 
    (VALUES
        ('supplier1', 'product1', 4),
        ('supplier1', 'product2', 3),
        ('supplier2', 'product3', 3),
        ('supplier2', 'product4', 4)
    )
AS Products(supplier_id, product_id, rating)
GROUP BY CUBE (supplier_id, rating, product_id);
--------------------------------------------------------------------------
SELECT supplier_id, rating, product_id, COUNT(*)
FROM 
    (VALUES
        ('supplier1', 'product1', 4),
        ('supplier1', 'product2', 3),
        ('supplier2', 'product3', 3),
        ('supplier2', 'product4', 4)
    )
AS Products(supplier_id, product_id, rating)
GROUP BY GROUPING SET (
        ( supplier_id, product_id, rating ),
        ( supplier_id, product_id         ),
        ( supplier_id,             rating ),
        ( supplier_id                     ),
        (              product_id, rating ),
        (              product_id         ),
        (                          rating ),
        (                                 )
    );

8.9.3. HAVING

HAVING消除不满足条件的组行。HAVING不同于WHERE:WHERE在GROUP BY之前过滤单独的行，而HAVING过滤GROUP BY创建的组行。HAVING条件引用的每个列必须是分组列中的列，以及聚合函数结果。

SELECT SUM(amount)
FROM Orders
GROUP BY users
HAVING SUM(amount) > 50;

HAVING的存在会将查询转换为分组查询，即使没有GROUP BY子句。这与查询包含聚合函数但没有GROUP BY子句时发生的情况相同。

查询会将所有选定的行组成一个组，SELECT列表和HAVING子句只能从聚合函数中引用列。如果HAVING条件为真，这样的查询将产生一行结果，如果不为真，则产生零行结果。

8.10. OVER聚合

流批处理任务均可使用。

OVER聚合会对输入的每一行有序数据计算聚合值。与GROUP BY聚合相比，OVER聚合不会将每个组的结果行数减少到一行。相反，OVER聚合为每个输入行生成一个聚合值。

下面的查询会为每个订单计算在当前订单之前一小时内收到的相同产品的所有订单的总和。

SELECT order_id, order_time, amount, 
    SUM(amount) OVER (
        PARTITION BY product
        ORDER BY order_time
        RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
    ) AS one_hour_prod_amount_sum
FROM Orders
;

OVER窗口的语法结构如下：

SELECT
    agg_func(agg_col) OVER (
        [PARTITION BY col1[, col2, ...]]
        ORDER BY time_col
        range_definition
    ),
    ...
FROM ...

可以在SELECT子句中定义多个OVER窗口聚合。但是，对于流查询，由于当前的限制，所有聚合的OVER窗口必须是相同的。

8.10.1. ORDER BY

OVER窗口定义在一个有序的行序列上。由于表数据没有固定的顺序，因此order by子句是强制的。对于流式查询，Flink目前只支持以升序时间属性列顺序定义的窗口。

8.10.2. PARTITION BY

可以在分区表上定义OVER窗口。如果存在PARTITION BY子句，则只在每个输入行所在分区的行上计算聚合。

8.10.3. Range定义

范围定义指定聚合中包含多少行。这个范围是用BETWEEN子句定义的，它定义了下边界和上边界。边界之间的所有行都会包含在聚合中。Flink只支持CURRENT ROW作为上边界。

有两个选项可以定义范围，ROWS间隔和range间隔。

RANGE intervals

RANGE间隔是在ORDER BY列的值上定义的，在Flink中，需要该列类型为时间属性。下面的RANGE间隔定义函数：聚合中包含时间为当前行30分钟的所有行。

RANGE BETWEEN INTERVAL '30' MINUTE PRECEDING AND CURRENT ROW

ROW intervals

ROWS间隔是一个基于计数的间隔。它确切地定义了聚合中包含的数据行数。下面的ROWS间隔定义函数：当前行和当前行之前的10行(总共11行)包含在聚合中。

ROWS BETWEEN 10 PRECEDING AND CURRENT ROW

8.10.4. WINDOW子句

WINDOW子句可用于在SELECT子句之外定义OVER窗口。它可以使查询更具可读性，也允许我们的多个聚合重用同一个窗口定义。

SELECT order_id, order_time, amount,
    SUM(amount) OVER w AS sum_amount,
    AVG(amount) OVER w AS avg_amount
FROM Orders
    WINDOW w AS (
        PARTITION BY product
        ORDER BY order_time
        RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
    )

8.11. Join

可同时用于流批处理任务。Flink SQL支持对动态表执行复杂而灵活的连接操作。有几种不同类型的连接来支持需要的各种查询。

默认情况下，表的连接顺序并不会优化查询效率。表是按照在FROM子句中指定的顺序连接的。通过先列出更新频率最低的表，然后列出更新频率最高的表，可以调整连接查询的性能。

确保以不会产生交叉连接(笛卡尔积)的顺序指定表即可，交叉连接不受支持，而且会导致查询失败。

8.11.1. 常规Join

常规连接是最通用的连接类型，其中任何新记录或对连接任意一侧的更改都会影响整个连接结果。例如，左表产生一条新记录，当产品id在右表可以找到时，它将与右表所有以前和将来的记录进行连接。

SELECT * FROM Orders
INNER JOIN Product
ON Orders.productId = Product.id;

对于流查询，常规连接的语法是最灵活的，可以使用任何类型的更新(插入、更新、删除)输入表。然而，该操作具有重要的含义：它要求连接输入的两张表永远在Flink中保持state状态。

因此，计算查询结果所需的状态可能会无限增长，这取决于所有输入表和中间连接结果的不同输入行数。可以适当配置查询的状态生存时间(TTL)，以防止状态大小过大。
注意，这可能会影响查询结果的正确性。详细信息请参见查询配置。

对于流查询，计算查询结果所需的状态可能会无限增长，这取决于聚合的类型和不同分组键的数量。请提供具有有效保留间隔的查询配置，以防止状态大小过大。

8.11.1.1. INNER等值连接

返回受连接条件限制的简单笛卡尔积。目前只支持等值连接，即至少具有一个具有相等谓词的连接条件的连接。不支持任意交叉或theta连接。

SELECT *
FROM Orders
INNER JOIN Product
ON Orders.product_id = Product.id;

8.11.1.2. OUTER等值连接

返回限定的笛卡尔积中的所有行(即，传递其连接条件的所有合并行)，加上连接条件与另一个表的任何行不匹配的外表中每一行的一个副本。

Flink支持左、右和全外连接。目前，只支持等值连接，即至少具有一个具有相等谓词的连接条件的连接。不支持任意交叉或theta连接。

SELECT *
FROM Orders
LEFT JOIN Product
ON Orders.product_id = Product.id;
--------------------------------------------------------------
SELECT *
FROM Orders
RIGHT JOIN Product
ON Orders.product_id = Product.id;
--------------------------------------------------------------
SELECT *
FROM Orders
FULL OUTER JOIN Product
ON Orders.product_id = Product.id;

8.11.2. Interval Join

返回受连接条件和时间约束限制的简单笛卡尔积。Interval Join需要至少一个等连接谓词和一个连接条件来限制双方的时间。

两个适当的范围谓词就可以定义这样的条件，比如：<、<=、>=、>、BETWEEN或单个相等谓词，都可以用于比较两个输入表的相同类型的时间属性(处理时间或事件时间)。

例如，如果在收到订单4小时后发货，则此查询将会把所有订单与其相应的发货关联起来。

SELECT *
FROM Orders o, Shipments s
WHERE o.id = s.order_id
    AND o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time;

以下谓词是有效的Interval Join条件的示例：

ltime = rtime
ltime >= rtime AND ltime < rtime + INTERVAL '10' MINUTE
ltime BETWEEN rtime - INTERVAL '10' SECOND AND rtime + INTERVAL '5' SECOND

流式连接查询与常规连接相比，Interval Join只支持带有时间属性的仅追加表。由于时间属性是准单调递增的，Flink可以在不影响结果正确性的情况下将旧值从其状态中移除。

8.11.3. Temporal Join

时态表是随着时间变化的表，在 flink 中也成为了动态表。时态表中的行在一个或多个时间周期内存在，所有Flink表都是时态的(动态的)。

时态表包含一个或多个版本表的快照，是一个可以追踪变更历史表（比如：数据库变更历史，包含所有的快照），或者是变化的维表（比如：包含最新快照的数据库表）。

8.11.3.1. Event Time Temporal Join

时态连接允许对版本化表进行连接，这意味着可以通过更改元数据来丰富表信息，并在某个时间点检索它的值。

时态连接取任意表(左输入/探查侧)，并将每一行与版本控制表(右输入/构建侧)中相应行的相关版本关联起来。Flink使用FOR SYSTEM_TIME AS of的SQL语法根据SQL:2011标准执行这个操作。时态连接的语法如下：

SELECT [column_list]
FROM table1 [AS <alias1>]
[LEFT] JOIN table2 FOR SYSTEM_TIME AS OF table1.{ proctime | rowtime } [AS <alias2>]
ON table1.column-name1 = table2.column-name1;

使用事件时间属性(即rowtime属性)，可以检索键在过去某个时刻的值。这允许在一个共同的时间点来连接两张表。版本化表将存储自最后一个水印以来所有版本的数据(按时间标识)。

例如，假设我们有一个订单表，每个订单的价格以不同的货币表示。要将此表适当地规范化为单一货币(如美元)，每个订单都需要与下单时的适当货币转换汇率连接起来。

-- 创建一个订单表，这是个标准的仅追加表。
CREATE TABLE orders (
    order_id STRING,
    price DECIMAL(32,2),
    currency STRING,
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time
) WITH (/* ... */);

-- 定义一个版本化表来存储货币转化率。这个表可以通过CDC定义，比如Debezium、压缩的kafka主题，或者是任何其他的方式定义版本化表。
CREATE TABLE currency_rates (
    currency STRING,
    conversion_rate DECIMAL(32, 2),
    update_time TIMESTAMP(3) METADATA FROM `values.source.timestamp` VIRTUAL
    WATERMARK FOR update_time AS update_time
) WITH (
    'connector' = 'upsert-kafka',
/* ... */
);

SELECT
    order_id,
    price,
    currency,
    conversion_rate,
    order_time,
FROM orders
LEFT JOIN currency_rates FOR SYSTEM_TIME AS OF orders.order_time
ON orders.currency = currency_rates.currency;
order_id  price  currency  conversion_rate  order_time
========  =====  ========  ===============  =========
o_001     11.11  EUR       1.14             12:00:00
o_002     12.51  EUR       1.10             12:06:00

注意：事件时间时态连接是由左右两边的水印触发的，连接的两张表都必须正确地设置水印。

注意：事件时间时态连接需要有包含主键的等值连接条件，例如，product_changelog表的主键P.product_id被约束在条件O.product_id = P.product_id中。

与常规连接相比，尽管构建端（右表）发生了更改，但前面的时态表结果并不会受到影响。与间隔连接（Interval Join）相比，时态表连接没有定义连接记录的时间窗口。

间隔连接包含时间窗口，时间窗口内的左右表数据都会进行连接。探测端（左表）记录总是在time属性指定的时间与构建端对应时间的数据进行连接。

因此，构建端的行可能是任意旧的。随着时间的推移，不再需要的记录版本(对于给定的主键)将从状态中删除。

8.11.3.2. Processing Time Temporal Join

处理时间时态表连接使用处理时间属性将行与外部版本表中键对应的最新版本数据进行关联。

根据定义，使用处理时间属性，连接将始终返回给定键的最新值。可以将查询表看作简单的HashMap<K, V>
，它存储了来自构建端的所有记录。这种连接的强大之处是，当不能在Flink中将表具体化为动态表时，它允许Flink直接针对外部系统工作。

下面的处理时间时态表连接示例显示了一个只追加的表订单，它与LatestRates表连接。LatestRates是一个维表(例如HBase表)，存储最新的比例。

在10:15,10:30,10:52,LatestRates的内容如下:

10:15> SELECT * FROM LatestRates;

currency   rate
======== ======
US Dollar   102
Euro        114
Yen           1

10:30> SELECT * FROM LatestRates;

currency   rate
======== ======
US Dollar   102
Euro        114
Yen           1

10:52> SELECT * FROM LatestRates;

currency   rate
======== ======
US Dollar   102
Euro        116     <==== changed from 114 to 116
Yen           1

10:15和10:30的LatestRates的内容是相等的。欧元汇率在10:52从114变到了116。

订单是一个仅追加表，表示给定金额和给定货币的支付数据。例如，在10:15有一个2欧元的订单。

SELECT * FROM Orders;

amount currency
====== =========
     2 Euro             <== arrived at time 10:15
     1 US Dollar        <== arrived at time 10:30
     2 Euro             <== arrived at time 10:52

根据这些表，来将所有订单转换为相同的货币。

amount currency     rate   amount*rate
====== ========= ======= ============
     2 Euro          114          228    <== arrived at time 10:15
     1 US Dollar     102          102    <== arrived at time 10:30
     2 Euro          116          232    <== arrived at time 10:52

在时态表连接的帮助下，我们可以在SQL中进行这样一个查询：

SELECT
o.amount, o.currency, r.rate, o.amount * r.rate
FROM Orders AS o
JOIN LatestRates FOR SYSTEM_TIME AS OF o.proctime AS r
ON r.currency = o.currency;

探测端（左表）中的每条记录都将与构建端表（右表）的当前版本记录进行连接。在上例中，使用了处理时间概念，因此在执行操作时，新添加的记录总是与最新版本的LatestRates表数据连接。

处理时间的结果是不确定的。处理时间时态连接最常使用外部表(即维度表)作为构建端（右表）。

与常规连接相比，尽管构建端（右表）发生了更改，前面的时态表结果也不会受到影响。与间隔连接相比，时态表连接没有定义记录连接的时间窗口，也就是说，旧行不会进行状态存储。

8.11.4. Lookup Join

Lookup Join通常使用从外部系统查询的数据来丰富表。连接要求一个表具有处理时间属性，另一个表由lookup source连接器支持。

查找连接使用上面的Processing Time Temporal join语法，并使用查找源连接器支持表。

下面的示例显示了指定Lookup Join的语法。

-- Customers通过JDBC连接器创建，并且可以被用于lookup joins
CREATE TEMPORARY TABLE Customers (
    id INT,
    name STRING,
    country STRING,
    zip STRING
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:mysql://mysqlhost:3306/customerdb',
    'table-name' = 'customers'
);

-- 通过customer来丰富订单信息
SELECT o.order_id, o.total, c.country, c.zip
FROM Orders AS o
JOIN Customers FOR SYSTEM_TIME AS OF o.proc_time AS c
ON o.customer_id = c.id;

在上面的示例中，Orders表使用来自MySQL数据库中的Customers表的数据进行数据信息扩展。通过后面处理时间属性的FOR SYSTEM_TIME AS OF子句确保在连接操作处理Orders行时，
Orders表的每一行都与那些匹配连接谓词的Customers行连接。它还防止在将来更新已连接的Customer行时更新连接结果。

Lookup Join还需要一个强制相等联接谓词，如上面示例中的o.customer_id = c.id。

8.11.5. Array展开

为给定数组中的每个元素返回新行。目前还不支持WITH ORDINALITY。

SELECT order_id, t.tag
FROM Orders
CROSS JOIN UNNEST(tags) AS t (tag);

8.11.6. 表函数

将表与表函数的结果进行连接。左表(外部表)的每一行都与对应的table函数调用产生的所有行连接。用户自定义的表函数在使用前必须注册。

8.11.6.1. INNER JOIN

如果左表(外部)的表函数调用返回空结果，则删除该行。

SELECT order_id, res
FROM Orders, LATERAL TABLE(table_func(order_id)) t(res);

8.11.6.2. LEFT OUTER JOIN

如果表函数调用返回空结果，则保留相应的左表数据行，并在结果中填充空值。目前，针对表的左外连接需要在ON子句中使用TRUE字面值。

SELECT order_id, res
FROM Orders
LEFT OUTER JOIN LATERAL TABLE(table_func(order_id)) t(res)
ON TRUE;

8.12. window join

从 fink-1.14.x 开始支持。

只能在流式处理中使用。

window join 会将时间尺度添加到他们的 join 标准中。如此一来， window join 就会将两个流中同一个窗口内拥有两个相同 key 的元素 join 起来。
window join 的语义和 DataStream window join 的语义相同。

对于 streaming 查询，和其他的流表 join 不同， window join 不会马上发射结果数据，而是在窗口结束之后发射最终的结果。此外，在不需要保存数据时， window join 会清除所有的中间状态数据。

通常来说，window join 会和窗口表值函数一起使用。因此，基于窗口表值函数，window join 后面可以接受其他函数操作，比如窗口聚合、窗口 TopN、window join。

目前，window join 要求 join 的输入表有相同的窗口开始和相同的窗口结束。

window join 支持 INNER、LEFT、RIGHT、FULL OUTER、ANTI、SEMI JOIN。

8.12.1. INNER/LEFT/RIGHT/FULL OUTER

下面的例子展示在 window join 语句中使用 INNER/LEFT/RIGHT/FULL OUTER 的语法。

SELECT ...
FROM L
[LEFT|RIGHT|FULL OUTER] JOIN R -- L 和 R 表可以是窗口表值函数产生的表
ON L.window_start = R.window_start AND L.window_end = R.window_end AND ...

INNER/LEFT/RIGHT/FULL OUTER WINDOW JOIN 的语法彼此是非常相似的，下面我们只给出了 FULL OUTER JOIN 的例子。

当使用 window join 时，在一个滚动窗口中，拥有相同 key 的所有元素将会 join 到一起。

下面的例子中，我们只在 window join 的滚动窗口上使用了一个滚动窗口表值函数。

通过将 join 的窗口限制在五分钟间隔，我们将会截断数据集为两个不同的窗口：[12:00, 12:05) 和 [12:05, 12:10)。L2 和 R2 行不会 join 到一起，因为他们处于不同的窗口中。

desc LeftTable;
+----------+------------------------+------+-----+--------+----------------------------------+
|     name |                   type | null | key | extras |                        watermark |
+----------+------------------------+------+-----+--------+----------------------------------+
| row_time | TIMESTAMP(3) *ROWTIME* | true |     |        | `row_time` - INTERVAL '1' SECOND |
|      num |                    INT | true |     |        |                                  |
|       id |                 STRING | true |     |        |                                  |
+----------+------------------------+------+-----+--------+----------------------------------+

SELECT * FROM LeftTable;
+------------------+-----+----+
|         row_time | num | id |
+------------------+-----+----+
| 2020-04-15 12:02 |   1 | L1 |
| 2020-04-15 12:06 |   2 | L2 |
| 2020-04-15 12:03 |   3 | L3 |
+------------------+-----+----+

desc RightTable;
+----------+------------------------+------+-----+--------+----------------------------------+
|     name |                   type | null | key | extras |                        watermark |
+----------+------------------------+------+-----+--------+----------------------------------+
| row_time | TIMESTAMP(3) *ROWTIME* | true |     |        | `row_time` - INTERVAL '1' SECOND |
|      num |                    INT | true |     |        |                                  |
|       id |                 STRING | true |     |        |                                  |
+----------+------------------------+------+-----+--------+----------------------------------+

SELECT * FROM RightTable;
+------------------+-----+----+
|         row_time | num | id |
+------------------+-----+----+
| 2020-04-15 12:01 |   2 | R2 |
| 2020-04-15 12:04 |   3 | R3 |
| 2020-04-15 12:05 |   4 | R4 |
+------------------+-----+----+
SELECT L.num as L_Num, L.id as L_Id, R.num as R_Num, R.id as R_Id, L.window_start, L.window_end
FROM
  (
  SELECT * FROM TABLE(TUMBLE(TABLE LeftTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
  ) L
FULL JOIN
  (
  SELECT * FROM TABLE(TUMBLE(TABLE RightTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
  ) R
ON L.num = R.num AND L.window_start = R.window_start AND L.window_end = R.window_end;
+-------+------+-------+------+------------------+------------------+
| L_Num | L_Id | R_Num | R_Id |     window_start |       window_end |
+-------+------+-------+------+------------------+------------------+
|     1 |   L1 |  null | null | 2020-04-15 12:00 | 2020-04-15 12:05 |
|  null | null |     2 |   R2 | 2020-04-15 12:00 | 2020-04-15 12:05 |
|     3 |   L3 |     3 |   R3 | 2020-04-15 12:00 | 2020-04-15 12:05 |
|     2 |   L2 |  null | null | 2020-04-15 12:05 | 2020-04-15 12:10 |
|  null | null |     4 |   R4 | 2020-04-15 12:05 | 2020-04-15 12:10 |
+-------+------+-------+------+------------------+------------------+

注意：为了更好的理解窗口 join 的行为，我们简单展示了时间戳值，并没有展示后面的0。
比如，TIMESTAMP(3) 类型的数据，在 FLINK CLI 中，2020-04-15 08:05 应该被展示为 2020-04-15 08:05:00.000 。

8.12.2. SEMI

在同一个窗口中，如果左表和右表至少有一行匹配，则 semi window join 会返回左表的一行记录。

SELECT *
FROM
    (
    SELECT * FROM TABLE(TUMBLE(TABLE LeftTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
    ) L
WHERE L.num IN
    (
    SELECT num
    FROM
        (   
        SELECT * FROM TABLE(TUMBLE(TABLE RightTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
        ) R
    WHERE L.window_start = R.window_start AND L.window_end = R.window_end
    );
+------------------+-----+----+------------------+------------------+-------------------------+
|         row_time | num | id |     window_start |       window_end |            window_time  |
+------------------+-----+----+------------------+------------------+-------------------------+
| 2020-04-15 12:03 |   3 | L3 | 2020-04-15 12:00 | 2020-04-15 12:05 | 2020-04-15 12:04:59.999 |
+------------------+-----+----+------------------+------------------+-------------------------+

SELECT *
FROM
    (
    SELECT * FROM TABLE(TUMBLE(TABLE LeftTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
    ) L
WHERE EXISTS
    (
    SELECT *
    FROM
        (
        SELECT * FROM TABLE(TUMBLE(TABLE RightTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
        ) R
    WHERE L.num = R.num AND L.window_start = R.window_start AND L.window_end = R.window_end
    );
+------------------+-----+----+------------------+------------------+-------------------------+
|         row_time | num | id |     window_start |       window_end |            window_time  |
+------------------+-----+----+------------------+------------------+-------------------------+
| 2020-04-15 12:03 |   3 | L3 | 2020-04-15 12:00 | 2020-04-15 12:05 | 2020-04-15 12:04:59.999 |
+------------------+-----+----+------------------+------------------+-------------------------+

8.12.3. ANTI

anti window join 会返回同一个窗口中所有没有 join 到一起的行。

SELECT *
FROM
    (
    SELECT * FROM TABLE(TUMBLE(TABLE LeftTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
    ) L
WHERE L.num NOT IN
    (
    SELECT num
    FROM
        (   
        SELECT * FROM TABLE(TUMBLE(TABLE RightTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
        ) R
    WHERE L.window_start = R.window_start AND L.window_end = R.window_end
    );
+------------------+-----+----+------------------+------------------+-------------------------+
|         row_time | num | id |     window_start |       window_end |            window_time  |
+------------------+-----+----+------------------+------------------+-------------------------+
| 2020-04-15 12:02 |   1 | L1 | 2020-04-15 12:00 | 2020-04-15 12:05 | 2020-04-15 12:04:59.999 |
| 2020-04-15 12:06 |   2 | L2 | 2020-04-15 12:05 | 2020-04-15 12:10 | 2020-04-15 12:09:59.999 |
+------------------+-----+----+------------------+------------------+-------------------------+

SELECT *
FROM
    (
    SELECT * FROM TABLE(TUMBLE(TABLE LeftTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
    ) L
WHERE NOT EXISTS
    (
    SELECT *
    FROM
        (
        SELECT * FROM TABLE(TUMBLE(TABLE RightTable, DESCRIPTOR(row_time), INTERVAL '5' MINUTES))
        ) R
    WHERE L.num = R.num AND L.window_start = R.window_start AND L.window_end = R.window_end
    );
+------------------+-----+----+------------------+------------------+-------------------------+
|         row_time | num | id |     window_start |       window_end |            window_time  |
+------------------+-----+----+------------------+------------------+-------------------------+
| 2020-04-15 12:02 |   1 | L1 | 2020-04-15 12:00 | 2020-04-15 12:05 | 2020-04-15 12:04:59.999 |
| 2020-04-15 12:06 |   2 | L2 | 2020-04-15 12:05 | 2020-04-15 12:10 | 2020-04-15 12:09:59.999 |
+------------------+-----+----+------------------+------------------+-------------------------+

8.12.4. 限制

8.12.4.1. join条件的限制

目前，window join 要求两张表拥有相同的窗口开始和窗口结束。未来，我们简化 join on 条件，如果窗口表值函数为滚动或者滑动，则只需要两张表有相同的窗口开始即可。

8.12.4.2. 在输入表上使用窗口TVF的限制

目前，左表和右表必须有相同的窗口表值函数。将来我们会扩展该功能，比如，有用相同窗口大小的滚动窗口 join 滑动窗口。

8.12.4.3. 直接在窗口TVF后使用window join的限制

目前，如果在窗口表值函数之后使用 window join ，窗口函数只能是滚动窗口、滑动窗口或者是累计窗口，而不能是会话窗口。

8.13. 集合操作

在流批任务中均可使用。

8.13.1. UNION

UNION和UNION ALL会返回两张表的所有行。UNION会对结果去重，而UNION ALL不会对结果行去重。

Flink SQL> create view t1(s) as values ('c'), ('a'), ('b'), ('b'), ('c');
Flink SQL> create view t2(s) as values ('d'), ('e'), ('a'), ('b'), ('b');
Flink SQL> (SELECT s FROM t1) UNION (SELECT s FROM t2);
+---+
| s|
+---+
| c|
| a|
| b|
| d|
| e|
+---+
Flink SQL> (SELECT s FROM t1) UNION ALL (SELECT s FROM t2);
+---+
| s|
+---+
| c|
| a|
| b|
| b|
| c|
| d|
| e|
| a|
| b|
| b|
+---+

8.13.2. INTERSECT

交集

INTERSECT和INTERSECT ALL返回在两个表中都存在的行。INTERSECT会对结果行去重，而INTERSECT ALL不会去重。

Flink SQL> (SELECT s FROM t1) INTERSECT (SELECT s FROM t2);
+---+
| s|
+---+
| a|
| b|
+---+
Flink SQL> (SELECT s FROM t1) INTERSECT ALL (SELECT s FROM t2);
+---+
| s|
+---+
| a|
| b|
| b|
+---+

8.13.3. EXCEPT

差集

EXCEPT和EXCEPT ALL返回在一个表中找到，但在另一个表中没有找到的行。EXCEPT会对结果去重，而EXCEPT ALL不会对结果去重。

Flink SQL> (SELECT s FROM t1) EXCEPT (SELECT s FROM t2);
+---+
| s |
+---+
| c |
+---+
Flink SQL> (SELECT s FROM t1) EXCEPT ALL (SELECT s FROM t2);
+---+
| s |
+---+
| c |
| c |
+---+

8.13.4. IN

如果外表字段值存在于给定的子查询结果表数据，则返回true。子查询结果表必须由一列组成。此列必须具有与表达式相同的数据类型。

SELECT user, amount
FROM Orders
WHERE product IN
    (
    SELECT product
    FROM NewProducts
    )
;

优化器将IN条件重写为join和group操作。对于流查询，计算查询结果所需的状态可能会无限增长，这取决于不同的输入行数。
可以通过配置合适的状态生存时间(TTL)，以防止状态大小过大。注意，这可能会影响查询结果的正确性。详细信息请参见查询配置。

8.13.5. EXISTS

SELECT user, amount
FROM Orders
WHERE product EXISTS
    (
    SELECT product
    FROM NewProducts
    )
;

如果子查询返回至少一行，则返回true。只有当操作可以在join和group操作中重写时才支持该语法。

优化器将EXISTS操作重写为join和group操作。对于流查询，计算查询结果所需的状态可能会无限增长，这取决于不同的输入行数。
可以通过配置合适的状态生存时间(TTL)，以防止状态大小过大。注意，这可能会影响查询结果的正确性。详细信息请参见查询配置。

8.14. ORDER BY子句

在流批任务中均可使用。

ORDER BY子句会根据指定的表达式对结果行进行排序。如果根据最左边的表达式比较，两行相等，则继续根据下一个表达式对它们进行比较，以此类推。如果根据所有指定的表达式比较，它们都是相等的，则以依赖于实现的顺序返回它们。

当以流模式运行时，表的主要排序顺序必须根据时间属性进行升序进行排序。所有后续排序都可以自由选择。但是在批处理模式中没有这种限制。

SELECT *
FROM Orders
ORDER BY order_time, order_id;

8.14.1. LIMIT子句

只能在批处理任务中使用。 LIMIT子句限制SELECT语句返回的行数。LIMIT通常与ORDER BY一起使用，以确保结果的确定性。

下面的示例返回Orders表中的前3行。

SELECT *
FROM Orders
ORDER BY orderTime
LIMIT 3;

8.15. Top-N

在流批任务中均可使用。

Top-N查询返回按列排序的N个最小或最大值。最小和最大的值集都被认为是Top-N查询。top-N查询在需要只显示批处理/流表中最下面的N条或最上面的N条记录的情况下非常有用。此结果集可用于进一步分析。

Flink使用OVER窗口子句和筛选条件的组合来表示Top-N查询。通过OVER窗口PARTITION BY子句的功能，Flink还支持多组Top-N。

例如，每个类别中实时销售额最高的前五种产品。对于批处理表和流表上的SQL，都支持Top-N查询。

Top-N语句的语法如下:

SELECT [column_list]
FROM
    (
    SELECT [column_list],
        ROW_NUMBER() OVER (
            [PARTITION BY col1[, col2...]]
            ORDER BY col1 [asc|desc][, col2 [asc|desc]...]
        ) AS rownum
    FROM table_name
    )
WHERE rownum <= N [AND conditions]

参数说明：

ROW_NUMBER()：根据分区中的数据行顺序，从1开始为每一行分配一个唯一的连续编号。目前，我们只支持ROW_NUMBER作为over window函数。将来，我们会支持RANK()和DENSE_RANK()。
PARTITION BY col1[, col2…]：分区列。每个分区都有一个Top-N结果。
ORDER BY col1 [asc|desc][, col2 [asc|desc]…]：指定排序列。不同列的排序方式可能不同。
WHERE rownum <= N： Flink需要rownum <= N来识别这个查询是Top-N查询。N表示将保留N条最小或最大的记录。
[AND conditions]：在where子句中可以随意添加其他条件，但其他条件只能使用AND关键字与rownum <= N组合。

注意：必须完全遵循上述模式，否则优化器将无法转换查询。

TopN查询结果为“结果更新”。Flink SQL将根据order键对输入数据流进行排序，因此如果前N条记录被更改，则更改后的记录将作为撤销/更新记录发送到下游。建议使用支持更新的存储作为Top-N查询的sink。

此外，如果top N记录需要存储在外部存储中，结果表应该具有与top-N查询相同的唯一键。

Top-N查询的唯一键是partition列和rownum列的组合。以下面的作业为例，假设product_id是ShopSales的唯一键，那么Top-N查询的唯一键是[category, rownum]和[product_id]。

下面的示例展示如何在流表上使用Top-N指定SQL查询。这是一个获得“每个类别中实时销售额最高的前五种产品”的例子。

CREATE TABLE ShopSales (
    product_id STRING,
    category STRING,
    product_name STRING,
    sales BIGINT
) WITH (...);

SELECT product_id, category, product_name, sales, row_num
FROM
    (
    SELECT product_id, category, product_name, sales,
        ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales DESC) AS row_num
    FROM ShopSales
    )
WHERE row_num <= 5;

无等级字段输出优化

如前所述，rownum字段将作为唯一键的一个字段写入结果表，这可能会导致大量记录被写入结果表。例如，当排名9的记录(比如product-1001)被更新并将其排名升级为1时，排名1 ~9的所有记录将作为更新消息输出到结果表。

如果结果表接收的数据过多，将成为SQL作业的瓶颈。优化方法是在Top-N查询的外部SELECT子句中省略rownum字段。这是合理的，因为前N个记录的数量通常不大，因此消费者可以自己快速地对结果记录进行排序。

在上面的示例中，如果没有rownum字段，只需要将更改后的记录(product-1001)发送到下游，这可以减少对结果表的大量IO。

下面的例子展示了如何用这种方式优化上面的Top-N：

CREATE TABLE ShopSales (
    product_id STRING,
    category STRING,
    product_name STRING,
    sales BIGINT
) WITH (...);

-- 输出时省略row_num字段
SELECT product_id, category, product_name, sales
FROM
    (
    SELECT product_id, category, product_name, sales,
        ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales DESC) AS row_num
    FROM ShopSales
    )
WHERE row_num <= 5;

注意在流模式下，为了将上述查询输出到外部存储并得到正确的结果，外部存储必须与Top-N查询具有相同的唯一键。在上面的查询示例中，如果product_id是查询的唯一键，那么外部表也应该将product_id作为唯一键。

8.16. Window Top-N

flink-1.13.x：只能在流模式任务中使用。
flink-1.15.x：可以在流批模式中使用。

Window Top-N是一个特殊的Top-N，它返回每个窗口以及其他分区键的N个最小或最大值。

对于流查询，与连续表上的常规top-N不同，窗口top-N不会发出中间结果，而只发出最终结果，即窗口末端的top-N所有记录。

此外，当不再需要时，窗口Top-N会清除所有中间状态。因此，如果用户不需要对每条记录更新结果，那么窗口Top-N查询会具有更好的性能。

flink-1.13.x：通常，窗口top-N与窗口聚合函数一起使用。
flink-1.14.x：通常，窗口 top-N 可以在窗口表值函数上直接使用，因此，窗口 top-N 可以基于窗口表值函数和其他函数一起使用，比如：窗口聚合、窗口 topN 和窗口 join。

Window Top-N可以用与常规Top-N相同的语法定义，此外，Window Top-N要求PARTITION BY子句包含window_start和window_end列，通过Windowing TVF或窗口聚合产生。

否则，优化器将无法翻译对应的sql查询。

Window Top-N语句的语法如下所示：

SELECT [column_list]
FROM
    (
    SELECT [column_list],
        ROW_NUMBER() OVER (
            PARTITION BY window_start, window_end [, col_key1...]
            ORDER BY col1 [asc|desc][, col2 [asc|desc]...]
        ) AS rownum
    FROM table_name
    ) -- 通过windowing TVF产生表
WHERE rownum <= N [AND conditions]

8.16.1. 在窗口聚合函数后使用窗口top-N

下面的例子展示如何计算每10分钟滚动窗口中销售额最高的前3个供应商：

-- 表必须有时间属性，比如下表中的bidtime列
Flink SQL> desc Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
| supplier_id |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

Flink SQL> SELECT * FROM Bid;
+------------------+-------+------+-------------+
| bidtime | price | item | supplier_id |
+------------------+-------+------+-------------+
| 2020-04-15 08:05 | 4.00 | A | supplier1 |
| 2020-04-15 08:06 | 4.00 | C | supplier2 |
| 2020-04-15 08:07 | 2.00 | G | supplier1 |
| 2020-04-15 08:08 | 2.00 | B | supplier3 |
| 2020-04-15 08:09 | 5.00 | D | supplier4 |
| 2020-04-15 08:11 | 2.00 | B | supplier3 |
| 2020-04-15 08:13 | 1.00 | E | supplier1 |
| 2020-04-15 08:15 | 3.00 | H | supplier2 |
| 2020-04-15 08:17 | 6.00 | F | supplier5 |
+------------------+-------+------+-------------+

Flink SQL> SELECT *
FROM
    (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY price DESC) as rownum
    FROM
        (
        SELECT window_start, window_end, supplier_id, SUM(price) as price, COUNT(*) as cnt
        FROM TABLE (TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES))
        GROUP BY window_start, window_end, supplier_id
        )
    )
WHERE rownum <= 3;
+------------------+------------------+-------------+-------+-----+--------+
| window_start | window_end | supplier_id | price | cnt | rownum |
+------------------+------------------+-------------+-------+-----+--------+
| 2020-04-15 08:00 | 2020-04-15 08:10 | supplier1 | 6.00 | 2 | 1 |
| 2020-04-15 08:00 | 2020-04-15 08:10 | supplier4 | 5.00 | 1 | 2 |
| 2020-04-15 08:00 | 2020-04-15 08:10 | supplier2 | 4.00 | 1 | 3 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | supplier5 | 6.00 | 1 | 1 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | supplier2 | 3.00 | 1 | 2 |
| 2020-04-15 08:10 | 2020-04-15 08:20 | supplier3 | 2.00 | 1 | 3 |
+------------------+------------------+-------------+-------+-----+--------+

注意:为了更好地理解窗口的行为，我们简化了时间戳值的显示，不显示秒小数点后的零。
例如，如果类型是timestamp(3)，在Flink SQL Client中，2020-04-15 08:05应该显示为2020-04-15 08:05:00.000。

8.16.2. 在窗口表值函数后使用窗口top-N

flink-1.14.x开始支持。

下面的例子展示怎么每个滚动窗口内 Top 3 的最高价格的 item。

SELECT *
FROM
    (
    SELECT bidtime, price, item, supplier_id, window_start, window_end, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY price DESC) as rownum
    FROM TABLE
        (
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10' MINUTES)
        )
    )
WHERE rownum <= 3;
+------------------+-------+------+-------------+------------------+------------------+--------+
|          bidtime | price | item | supplier_id |     window_start |       window_end | rownum |
+------------------+-------+------+-------------+------------------+------------------+--------+
| 2020-04-15 08:05 |  4.00 |    A |   supplier1 | 2020-04-15 08:00 | 2020-04-15 08:10 |      2 |
| 2020-04-15 08:06 |  4.00 |    C |   supplier2 | 2020-04-15 08:00 | 2020-04-15 08:10 |      3 |
| 2020-04-15 08:09 |  5.00 |    D |   supplier4 | 2020-04-15 08:00 | 2020-04-15 08:10 |      1 |
| 2020-04-15 08:11 |  2.00 |    B |   supplier3 | 2020-04-15 08:10 | 2020-04-15 08:20 |      3 |
| 2020-04-15 08:15 |  3.00 |    H |   supplier2 | 2020-04-15 08:10 | 2020-04-15 08:20 |      2 |
| 2020-04-15 08:17 |  6.00 |    F |   supplier5 | 2020-04-15 08:10 | 2020-04-15 08:20 |      1 |
+------------------+-------+------+-------------+------------------+------------------+--------+

注意:为了更好地理解窗口的行为，我们简化了时间戳值的显示，不显示秒小数点后的零，例如，如果类型是timestamp(3)，在Flink SQL Client中，2020-04-15 08:05应该显示为2020-04-15 08:05:
00.000。

8.16.3. 限制

flink-1.13.x：目前，Flink只支持Window Top-N紧随Window Aggregation产生的表。在不久的将来，将支持Window TVF之后的Window Top-N。
flink-1.14.x、flink-1.15.x：目前，flink只支持在滚动窗口、滑动窗口、累加窗口这三个窗口表值函数之后直接使用 window Top-N。
在不久的将来，将支持在会话窗口的窗口表值函数之后直接使用 window Top-N 。

8.17. 去重

在流批模式中均可使用。

去重会删除在一组列上重复的行，只保留第一行或最后一行。在某些情况下，上游ETL作业并不是端到端精确一次的；当发生故障转移时，这可能会导致接收器中出现重复记录。

重复记录会影响下游分析作业(如SUM、COUNT)的正确性，因此需要在进一步分析之前进行重复数据删除。

Flink使用ROW_NUMBER()来删除重复数据，就像Top-N查询一样。理论上，重复数据删除是Top-N的一种特殊情况，其中N为1，按处理时间或事件时间排序。

重复数据删除语句的语法如下：

SELECT [column_list]
FROM
    (
    SELECT [column_list],
        ROW_NUMBER() OVER (
            [PARTITION BY col1[, col2...]]
            ORDER BY time_attr [asc|desc]
        ) AS rownum
    FROM table_name
    )
WHERE rownum = 1

参数说明：

ROW_NUMBER()：为每一行分配一个唯一的连续编号，从1开始。
PARTITION BY col1[, col2…]：指定分区列，即重复数据删除键。
ORDER BY time_attr [asc|desc]：排序列，必须是时间属性。目前Flink支持处理时间属性和事件时间属性。按ASC排序意味着保留第一行，按DESC排序意味着保留最后一行。
WHERE rownum = 1：Flink需要rownum = 1来识别这个查询是重复数据删除。

注意：必须完全遵循上述模式，否则优化器将无法转换查询。

以下示例展示如何在流表上使用重复数据删除的SQL语句：

CREATE TABLE Orders (
    order_time STRING,
    user STRING,
    product STRING,
    num BIGINT,
    proctime AS PROCTIME()
) WITH (...);
-- 移除重复的order_id行数据，只保留第一个接收到的行数据，因为同一个order_id不应该出现两个订单
SELECT order_id, user, product, num
FROM
    (
    SELECT *,
        ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY proctime ASC) AS row_num
    FROM Orders
    )
WHERE row_num = 1;

8.18. 窗口去重

flink-1.15.x开始支持。

窗口去重是一种特殊的去重，它会删除数据集中重复的行，保留每个窗口和分区中 key 对应的第一行或者是最后一行。

对于流式查询，和常规的流表去重不同，窗口去重并不会马上发射结果数据，而是在窗口结束时发射最终的结果。另外，窗口去重会在不需要时删除所有的中间状态数据。
因此，如果用户不需要实时使用结果数据更新每一条记录，窗口去重会有更好的性能。
通常，窗口去重会在窗口表值函数上直接使用。因此，窗口去重应该基于窗口表值函数并且和其他函数一起使用，比如：窗口聚合函数、窗口 TopN、窗口 join。

窗口去重定义和常规去重定义语法一样，具体可查看去重文档。除此之外，窗口去重要求 PARTITION BY 语句包含 window_start 和 window_end 字段。否则，优化器无法翻译该查询。

flink 使用 ROW_NUMBER() 来去掉重复数据，和窗口 Top-N 方式一样。从理论上来说，窗口去重是窗口 Top-N 的一种特殊用例，要求 N 为1，并且按照处理时间或者事件时间排序。

下例展示窗口去重语句的语法：

SELECT [column_list]
FROM
(
SELECT [column_list],
ROW_NUMBER() OVER (
PARTITION BY window_start, window_end [, col_key1...]
ORDER BY time_attr [asc|desc]
) AS rownum
FROM table_name
) -- 该关系接收窗口表值函数
WHERE (rownum = 1 | rownum <=1 | rownum < 2) [AND conditions]

参数说明：

ROW_NUMBER()：给每一行分配一个唯一、特殊的数字，从1开始。
PARTITION BY window_start, window_end [, col_key1…]：指定分区字段，包含：window_start、window_end 和其他分区 key。
ORDER BY time_attr [asc|desc]：指定排序字段，该字段必须是时间类型。目前 flink 支持处理时间和事件时间。 Ordering by ASC 表示保留第一行，ordering by DESC
表示保留最后一行。
WHERE (rownum = 1 | rownum <=1 | rownum < 2)：必须有 rownum = 1 | rownum <=1 | rownum < 2 ，这是为了让优化器取失败该查询，并且将其翻译为窗口去重。

注意：sql 语句必须完全匹配上述模式，否则优化器无法将查询翻译为窗口去重。

8.18.1. 案例

下例展示怎么去保留10分钟的滚动窗口的最后一行数据。

-- 表必须有时间字段，比如该表中的`bidtime`字段。
Flink SQL> DESC Bid;
+-------------+------------------------+------+-----+--------+---------------------------------+
|        name |                   type | null | key | extras |                       watermark |
+-------------+------------------------+------+-----+--------+---------------------------------+
|     bidtime | TIMESTAMP(3) *ROWTIME* | true |     |        | `bidtime` - INTERVAL '1' SECOND |
|       price |         DECIMAL(10, 2) | true |     |        |                                 |
|        item |                 STRING | true |     |        |                                 |
+-------------+------------------------+------+-----+--------+---------------------------------+

Flink SQL> SELECT * FROM Bid;
+------------------+-------+------+
|          bidtime | price | item |
+------------------+-------+------+
| 2020-04-15 08:05 |  4.00 | C    |
| 2020-04-15 08:07 |  2.00 | A    |
| 2020-04-15 08:09 |  5.00 | D    |
| 2020-04-15 08:11 |  3.00 | B    |
| 2020-04-15 08:13 |  1.00 | E    |
| 2020-04-15 08:17 |  6.00 | F    |
+------------------+-------+------+

SELECT *
FROM (
    SELECT bidtime, price, item, supplier_id, window_start, window_end,
        ROW_NUMBER() OVER (
            PARTITION BY window_start, window_end
            ORDER BY bidtime DESC
        ) AS rownum
    FROM TABLE(
        TUMBLE(TABLE Bid, DESCRIPTOR(bidtime), INTERVAL '10'

9. DESCRIBE

9.1. 介绍

DESCRIBE语句用于描述表或视图的schema。

9.2. 语法

{ DESCRIBE | DESC } [catalog_name.][db_name.]table_name

10. ALTER

10.1. 介绍

ALTER语句用于修改Catalog中已注册的表/视图/函数定义。

Flink SQL目前支持以下ALTER语句：

ALTER TABLE
ALTER VIEW（从1.14.x开始支持）
ALTER DATABASE
ALTER FUNCTION

10.1.1. ALTER TABLE

重命名表

ALTER TABLE [catalog_name.][db_name.]table_name RENAME TO new_table_name

将给定的表名重命名为另一个新表名。设置或更改表属性:

ALTER TABLE [catalog_name.][db_name.]table_name SET (key1=val1, key2=val2, ...)

给指定的表设置一个或多个属性。如果表中已经设置了特定的属性，则用新值覆盖旧值。

10.1.2. ALTER VIEW

从1.14.x开始支持。

ALTER VIEW [catalog_name.][db_name.]view_name RENAME TO new_view_name

将之前 catalog 和 database 下的视图重名为新的名称。

ALTER VIEW [catalog_name.][db_name.]view_name AS new_query_expression

改变视图之前的查询定义为新的查询。

10.1.3. ALTER DATABASE

ALTER DATABASE [catalog_name.]db_name SET (key1=val1, key2=val2, ...)

给指定的数据库设置一个或多个属性。如果数据库中已经设置了特定的属性，则使用新值覆盖旧值。

10.1.4. ALTER FUNCTION

ALTER [TEMPORARY|TEMPORARY SYSTEM] FUNCTION [IF EXISTS] [catalog_name.][db_name.]function_name AS identifier [LANGUAGE JAVA|SCALA|PYTHON]

使用新的标识符和可选的语言标记更改catalog函数。如果函数在catalog中不存在，则抛出异常。如果语言标记是JAVA/SCALA，则标识符是UDF的完整类路径。关于Java/Scala UDF的实现，请参考用户自定义函数。

TEMPORARY

更改具有catalog和数据库名称空间的临时catalog函数，并重写catalog函数。

TEMPORARY SYSTEM

更改没有名称空间的临时系统函数并覆盖内置函数。

IF EXISTS

如果函数不存在，什么也不会发生。

LANGUAGE JAVA|SCALA|PYTHON

用于指示flink运行时如何执行该函数的语言标记。目前只支持JAVA、SCALA和PYTHON，函数默认语言为JAVA。

11. EXPLAIN

11.1. 介绍

EXPLAIN语句用于解释SELECT或INSERT语句的逻辑和优化的查询计划。

11.2. 语法

flink-1.13.x：

EXPLAIN PLAN FOR <query_statement_or_insert_statement>

flink-1.14.x：

EXPLAIN [([ExplainDetail[, ExplainDetail]*]) | PLAN FOR] <query_statement_or_insert_statement>

11.3. 案例

Flink SQL> CREATE TABLE MyTable1 (`count` bigint, word VARCHAR(256)) WITH (''connector'' = ''datagen'');
[INFO] Table has been created.

Flink SQL> CREATE TABLE MyTable2 (`count` bigint, word VARCHAR(256)) WITH (''connector'' = ''datagen'');
[INFO] Table has been created.

Flink SQL> EXPLAIN PLAN FOR SELECT `count`, word FROM MyTable1 WHERE word LIKE ''F%''
> UNION ALL
> SELECT `count`, word FROM MyTable2;
-- 下面的语法从1.14.x开始支持
Flink SQL> EXPLAIN ESTIMATED_COST, CHANGELOG_MODE, JSON_EXECUTION_PLAN SELECT `count`, word FROM MyTable1
> WHERE word LIKE ''F%''
> UNION ALL
> SELECT `count`, word FROM MyTable2;

11.4. 结果

11.4.1. EXPLAIN PLAN

== Abstract Syntax Tree ==
LogicalUnion(all=[true])
:- LogicalProject(count=[$0], word=[$1])
:  +- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
:     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
+- LogicalProject(count=[$0], word=[$1])
   +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])

== Optimized Physical Plan ==
Union(all=[true], union=[count, word])
:- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
:  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])

== Optimized Execution Plan ==
Union(all=[true], union=[count, word])
:- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
:  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])

11.4.2. EXPLAIN PLAN WITH DETAILS

== Abstract Syntax Tree ==
LogicalUnion(all=[true])
:- LogicalProject(count=[$0], word=[$1])
:  +- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
:     +- LogicalTableScan(table=[[default_catalog, default_database, MyTable1]])
+- LogicalProject(count=[$0], word=[$1])
   +- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])

== Optimized Physical Plan ==
Union(all=[true], union=[count, word], changelogMode=[I]): rowcount = 1.05E8, cumulative cost = {3.1E8 rows, 3.05E8 cpu, 4.0E9 io, 0.0 network, 0.0 memory}
:- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')], changelogMode=[I]): rowcount = 5000000.0, cumulative cost = {1.05E8 rows, 1.0E8 cpu, 2.0E9 io, 0.0 network, 0.0 memory}
:  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word], changelogMode=[I]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 2.0E9 io, 0.0 network, 0.0 memory}
+- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word], changelogMode=[I]): rowcount = 1.0E8, cumulative cost = {1.0E8 rows, 1.0E8 cpu, 2.0E9 io, 0.0 network, 0.0 memory}

== Optimized Execution Plan ==
Union(all=[true], union=[count, word])
:- Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
:  +- TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])

== Physical Execution Plan ==
{
  "nodes" : [ {
    "id" : 37,
    "type" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])",
    "pact" : "Data Source",
    "contents" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable1]], fields=[count, word])",
    "parallelism" : 1
  }, {
    "id" : 38,
    "type" : "Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])",
    "pact" : "Operator",
    "contents" : "Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])",
    "parallelism" : 1,
    "predecessors" : [ {
      "id" : 37,
      "ship_strategy" : "FORWARD",
      "side" : "second"
    } ]
  }, {
    "id" : 39,
    "type" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])",
    "pact" : "Data Source",
    "contents" : "Source: TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[count, word])",
    "parallelism" : 1
  } ]

11.5. Explain细节

从flink-1.14.x开始支持。

打印语句包含指定explain细节的 plan 信息。

ESTIMATED_COST：估计成本，生成优化器估计的物理节点的成本信息，
比如：TableSourceScan(…, cumulative cost ={1.0E8 rows, 1.0E8 cpu, 2.4E9 io, 0.0 network, 0.0 memory})。
CHANGELOG_MODE：为每个物理节点生成变更日志模式，比如：GroupAggregate(…, changelogMode=[I,UA,D])。
JSON_EXECUTION_PLAN：生成json格式的程序执行计划。

12. DROP

12.1. 介绍

DROP语句用于从当前或指定的Catalog中删除已注册的表/视图/函数。 Flink SQL目前支持以下DROP语句：

DROP CATALOG（从1.14.x开始支持）
DROP TABLE
DROP DATABASE
DROP VIEW
DROP FUNCTION

12.1.1. DROP CATALOG

从1.14.x开始支持。

DROP CATALOG [IF EXISTS] catalog_name

删除指定的 catalog。

IF EXISTS

如果该 catalog 不存在，则什么也不会发生。

12.1.2. DROP TABLE

DROP [TEMPORARY] TABLE [IF EXISTS] [catalog_name.][db_name.]table_name

删除指定表名的表。如果要删除的表不存在，则抛出异常。

TEMPORARY

删除具有目录和数据库名称空间的临时表。

IF EXISTS

如果该表不存在，则什么也不会发生。

12.1.3. DROP DATABASE

DROP DATABASE [IF EXISTS] [catalog_name.]db_name [ (RESTRICT | CASCADE) ]

删除给定数据库名称的数据库。如果要删除的数据库不存在，则抛出异常。

IF EXISTS

如果数据库不存在，则什么也不会发生。

RESTRICT

删除非空数据库将触发异常。默认启用。

CASCADE

删除非空数据库时也会删除所有相关的表和函数。

12.1.4. DROP VIEW

DROP [TEMPORARY] VIEW  [IF EXISTS] [catalog_name.][db_name.]view_name

删除具有目录和数据库名称空间的视图。如果要删除的视图不存在，则抛出异常。

TEMPORARY

删除具有目录和数据库名称空间的临时视图。

IF EXISTS

如果视图不存在，则什么也不会发生。Flink不通过CASCADE/RESTRICT关键字维护视图的依赖关系，当前的方式是当用户试图在视图的底层表被删除的情况下使用视图时抛出延迟错误消息。

12.1.5. DROP FUNCTION

DROP [TEMPORARY|TEMPORARY SYSTEM] FUNCTION [IF EXISTS] [catalog_name.][db_name.]function_name

删除具有目录和数据库名称空间的目录函数。如果要删除的函数不存在，则抛出异常。

TEMPORARY

删除具有目录和数据库名称空间的临时目录函数。

TEMPORARY SYSTEM

删除没有命名空间的临时系统函数。

IF EXISTS

如果函数不存在，什么也不会发生。

13. LOAD

13.1. 介绍

LOAD语句用于加载内置或用户自定义的模块。

13.2. LOAD模块

语法结构：

LOAD MODULE module_name [WITH ('key1' = 'val1', 'key2' = 'val2', ...)]

module_name是一个简单的标识符，区分大小写。它应该与模块工厂中定义的模块类型相同，其用于模块的发现。

properties (‘key1’ = ‘val1’， ‘key2’ = ‘val2’，…)是一个映射，包含一组键值对(除了key ‘type’)，传递给发现服务相对应的模块。

13.3. 案例代码

-- 加载 hive 模块
load module hive with ('hive-version' = '2.3.6');
-- 推荐下面这种写法，不指定 hive 的版本，由系统去自动提取。
load module hive;

给平台添加了 flink-connector-sql-hive 依赖之后，就相当于已经添加了 hive 模块的实现，因此可以直接去加载 hive 模块。

在 flink sql 中加载了hive模块，并且use了hive模块之后，查询语句中就可以直接去使用hive中的函数了。

14. UNLOAD

14.1. 介绍

UNLOAD语句用于卸载内置或用户自定义的模块。

14.2. UNLOAD模块

语法结构：

UNLOAD MODULE module_name

15. SHOW

15.1. 介绍

SHOW语句用于列出所有catalog，或在当前catalog，列出所有数据库中所有表/视图/字段，或列出当前catalog和当前数据库，或显示当前catalog和数据库，
或列出当前catalog和当前的数据库的所有函数包括系统函数和用户自定义的函数。或只列出当前catalog和当前数据库中用户自定义的函数，或列出启用的模块名，或列出当前会话中为启用状态的所有已加载模块。

SHOW CREATE 语句被用于打印 DDL 语句，目前， SHOW CREATE 语句值能用于打印给定表或试图的 DDL 语句。 Flink SQL目前支持以下SHOW语句：

SHOW CATALOGS
SHOW CURRENT CATALOG
SHOW DATABASES
SHOW CURRENT DATABASE
SHOW TABLES
SHOW CREATE TABLE（从1.14.x开始支持）
SHOW COLUMNS（从1.15.x开始支持）
SHOW VIEWS
SHOW CREATE VIEW（从1.15.x开始支持）
SHOW FUNCTIONS
SHOW MODULES

15.2. SHOW CATALOGS

SHOW CATALOGS

显示所有catalog。

15.3. SHOW CURRENT CATALOG

SHOW CURRENT CATALOG

展示当前catalog。

15.4. SHOW DATABASES

SHOW DATABASES

展示当前catalog里的所有数据库。

15.5. SHOW CURRENT DATABASE

SHOW CURRENT DATABASE

展示当前数据库。

15.6. SHOW TABLES

15.6.1. flink-1.13.x

SHOW TABLES

展示当前catalog和当前数据库中的所有表。

15.6.2. flink-1.15.x

SHOW TABLES [ ( FROM | IN ) [catalog_name.]database_name ] [ [NOT] LIKE <sql_like_pattern> ]

展示可选的指定数据库中的所有表。如果没有指定数据库，则返回当前数据库的表。另外，该语句可以通过一个可选的匹配表达式对表进行过滤。

LIKE：展示通过 like 关键字指定名称的表，like 语法和 <sql_like_pahhern> 类似。LIKE 关键字的语法和 MySQL 方言一样：

% 匹配任意个数的字符，包括零个字符， % 匹配一个 % 符号。
_ 只匹配一个字符， _ 匹配一个 _ 符号。

15.6.2.1. SHOW TABLES案例

从1.15.x开始支持。

假设名为 catalog1 的 catalog 下的 db1 数据库并且包含以下表：

person
dim

当前会话中的数据库包含以下表：

fights
orders

展示给定数据库下的所有表：

show tables from db1;
-- show tables from catalog1.db1;
-- show tables in db1;
-- show tables in catalog1.db1;
+------------+
| table name |
+------------+
|        dim |
|     person |
+------------+
2 rows in set

通过给定的 sql 匹配，展示给定数据库下的表：

show tables from db1 like '%n';
-- show tables from catalog1.db1 like '%n';
-- show tables in db1 like '%n';
-- show tables in catalog1.db1 like '%n';
+------------+
| table name |
+------------+
| person |
+------------+
1 row in set

通过给定的 sql 匹配，展示给定数据库下不符合 sql 匹配的表：

show tables from db1 not like '%n';
-- show tables from catalog1.db1 not like '%n';
-- show tables in db1 not like '%n';
-- show tables in catalog1.db1 not like '%n';
+------------+
| table name |
+------------+
| dim |
+------------+
1 row in set

展示当前数据库下的所有表：

show tables;
+------------+
| table name |
+------------+
| items |
| orders |
+------------+
2 rows in set

15.7. SHOW CREATE TABLE

从1.14.x开始支持。

SHOW CREATE TABLE

展示指定表的建表语句。

另外：当前的 SHOW CREATE TABLE 语句只支持展示通过 flink SQL DDL 创建的表。

15.8. SHOW COLUMNS

从1.15.x开始支持。

SHOW COLUMNS ( FROM | IN ) [[catalog_name.]database.]<table_name> [ [NOT] LIKE <sql_like_pattern>]

通过给定的表情和可选的 like 语句展示表的字段信息。

LIKE：展示通过 like 关键字指定名称的表的字段信息，like 语法和 <sql_like_pattern> 类似。

LIKE 关键字的语法和 MySQL 方言一样：

% 匹配任意个数的字符，包括零个字符， % 匹配一个 % 符号。
_ 只匹配一个字符， _ 匹配一个 _ 符号。

15.8.1. SHOW COLUMNS案例

从1.15.x开始支持。

假设名为 catalog1 的 catalog 下的 database1 数据库下的 orders 表有如下结构：

+---------+-----------------------------+-------+-----------+---------------+----------------------------+
|    name |                        type |  null |       key |        extras |                  watermark |
+---------+-----------------------------+-------+-----------+---------------+----------------------------+
|    user |                      BIGINT | false | PRI(user) |               |                            |
| product |                 VARCHAR(32) |  true |           |               |                            |
|  amount |                         INT |  true |           |               |                            |
|      ts |      TIMESTAMP(3) *ROWTIME* |  true |           |               | `ts` - INTERVAL '1' SECOND |
|   ptime | TIMESTAMP_LTZ(3) *PROCTIME* | false |           | AS PROCTIME() |                            |
+---------+-----------------------------+-------+-----------+---------------+----------------------------+

展示给定表的所有字段：

show columns from orders;
-- show columns from database1.orders;
-- show columns from catalog1.database1.orders;
-- show columns in orders;
-- show columns in database1.orders;
-- show columns in catalog1.database1.orders;
+---------+-----------------------------+-------+-----------+---------------+----------------------------+
|    name |                        type |  null |       key |        extras |                  watermark |
+---------+-----------------------------+-------+-----------+---------------+----------------------------+
|    user |                      BIGINT | false | PRI(user) |               |                            |
| product |                 VARCHAR(32) |  true |           |               |                            |
|  amount |                         INT |  true |           |               |                            |
|      ts |      TIMESTAMP(3) *ROWTIME* |  true |           |               | `ts` - INTERVAL '1' SECOND |
|   ptime | TIMESTAMP_LTZ(3) *PROCTIME* | false |           | AS PROCTIME() |                            |
+---------+-----------------------------+-------+-----------+---------------+----------------------------+
5 rows in set

展示表中符合给定 sql 匹配的字段信息：

show columns from orders like '%r';
-- show columns from database1.orders like '%r';
-- show columns from catalog1.database1.orders like '%r';
-- show columns in orders like '%r';
-- show columns in database1.orders like '%r';
-- show columns in catalog1.database1.orders like '%r';
+------+--------+-------+-----------+--------+-----------+
| name |   type |  null |       key | extras | watermark |
+------+--------+-------+-----------+--------+-----------+
| user | BIGINT | false | PRI(user) |        |           |
+------+--------+-------+-----------+--------+-----------+
1 row in set

展示表中不符合给定 sql 匹配的字段信息：

show columns from orders not like '%_r';
-- show columns from database1.orders not like '%_r';
-- show columns from catalog1.database1.orders not like '%_r';
-- show columns in orders not like '%_r';
-- show columns in database1.orders not like '%_r';
-- show columns in catalog1.database1.orders not like '%_r';
+---------+-----------------------------+-------+-----+---------------+----------------------------+
|    name |                        type |  null | key |        extras |                  watermark |
+---------+-----------------------------+-------+-----+---------------+----------------------------+
| product |                 VARCHAR(32) |  true |     |               |                            |
|  amount |                         INT |  true |     |               |                            |
|      ts |      TIMESTAMP(3) *ROWTIME* |  true |     |               | `ts` - INTERVAL '1' SECOND |
|   ptime | TIMESTAMP_LTZ(3) *PROCTIME* | false |     | AS PROCTIME() |                            |
+---------+-----------------------------+-------+-----+---------------+----------------------------+
4 rows in set

15.9. SHOW VIEWS

SHOW VIEWS

展示当前catalog和当前数据库中的所有视图。

15.10. SHOW CREATE VIEW

从1.15.x开始支持。

SHOW CREATE VIEW [catalog_name.][db_name.]view_name

展示给定视图的创建语句。

15.11. SHOW FUNCTIONS

SHOW [USER] FUNCTIONS

展示当前catalog和当前数据库中的所有系统和自定义函数。

USER

值展示当前catalog和当前数据库中的所有自定义函数。

15.12. SHOW MODULES

SHOW [FULL] MODULES

按解析顺序显示所有启用的模块名称。

FULL

按照顺序显示所有启用状态的已加载模块。

最后

深知大多数初中级Java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则近万的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《Java开发全套学习资料》送给大家，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

小编已加密：aHR0cHM6Ly9kb2NzLnFxLmNvbS9kb2MvRFVrVm9aSGxQZUVsTlkwUnc==出于安全原因，我们把网站通过base64编码了，大家可以通过base64解码把网址获取下来。

【图像分割】基于matlab萤火虫算法图像聚类分割【含Matlab源码 2106期】

SQL 入门之第一讲——MySQL 8.0.29安装教程(windows 64位)

用Python进行数学建模（一）

力扣：669. 修剪二叉搜索树，今日份快乐

java毕业设计KTV点歌系统mybatis+源码+调试部署+系统+数据库+lw

[python] 基于diagrams库绘制系统架构图

2022 年全国职业院校技能大赛（中职组）网络安全竞赛试题D模块评分标准

C++ 语法基础课1 —— 变量、输入输出、顺序语句

M的编程备忘录之C++——map和set

《Orange‘s 一个操作系统的实现》第六章

Spring Cloud 拉取 Nacos 中配置文件

python-中断time.sleep一种更优雅的办法:event.wait

【毕业设计】大数据公交数据分析与可视化 - 大数据 python falsk

Hadoop与Spark中的Shuffle过程梳理

CH9101芯片应用—硬件设计指南

【翻译】Mashape是如何管理15000个API和微服务的（三）

bootstrap创建登录注册页面

CSS实用技巧干货

iOS帅气加载动画、通知视图、红包助手、引导页、导航栏、朋友圈、小游戏等效果源码...

JavaScript设计模式系列一：工厂模式

Linux快速配置 VIM 实现语法高亮补全缩进等功能

Netty源码解析1-Buffer

Spring声明式事务管理之一：五大属性分析

TiDB 源码阅读系列文章（十）Chunk 和执行框架简介

阿里研究院入选中国企业智库系统影响力榜

不发不行！Netty集成文字图片聊天室外加TCP/IP软硬件通信

目录与文件属性：编写ls

前端性能优化——回流与重绘

人脸识别最新开发经验demo

体验javascript之美-第五课匿名函数自执行和闭包是一回事儿吗？

王永庆：技术创新改变教育未来

微服务入门【系列视频课程】

与 ConTeXt MkIV 官方文档的接驳

深度学习之轻量级神经网络在TWS蓝牙音频处理器上的部署

elasticsearch-head插件安装

hi-nginx-1.3.4编译安装

Spring Boot 分片上传文件

水经微图Web1.5.0版即将上线

#经典论文异质山坡的物理模型 2 有效导水率

#微信小程序（布局、渲染层基础知识）

（算法二）滑动窗口

（一）eclipse Dynamic web project 工程目录以及文件路径问题

（转)IIS6 ASP 0251超过响应缓冲区限制错误的解决方法

.NET 3.0 Framework已经被添加到WindowUpdate

.net core 6 使用注解自动注入实例，无需构造注入 autowrite4net

.Net6支持的操作系统版本(.net8已来，你还在用.netframework4.5吗)

.NET的微型Web框架 Nancy

@JSONField或@JsonProperty注解使用

[8-23]知识梳理：文件系统、Bash基础特性、目录管理、文件管理、文本查看编辑处理...

[Android]通过PhoneLookup读取所有电话号码

[AS3]URLLoader+URLRequest+JPGEncoder实现BitmapData图片数据保存

[C++]——带你学习类和对象

[CSS3备忘] transform animation 等

[Django 0-1] Core.Handlers 模块

[flink总结]什么是flink背压 ,有什么危害? 如何解决flink背压?flink如何保证端到端一致性?