主要引用官方文档 https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/table/sql/queries/overview/
SELECT 语句和 VALUES 语句需要使用 TableEnvironment 的 sqlQuery()
方法加以指定,会以 Table 的形式返回 SELECT (或 VALUE)的查询结果。Table 可被用于 SQL 或 Table API 查询、转换为 DataSet 或 DataStream、输出到 TableSink。SQL 与 Table API 的查询可以进行无缝融合、整体优化。
为了可以在 SQL 查询中访问到表,需要先在 TableEnvironment 中注册表(可以通过 TableSource、Table、CREATE TABLE 语句、DataStream 或 DataSet 注册)。为方便起见 Table.toString() 将会在其 TableEnvironment 中以唯一的名称自动注册表,并返回名称。
注意: 查询若包括了不支持的 SQL 特性,将会抛出 TableException
。
指定查询
以下示例显示如何在已注册和内联表上指定 SQL 查询。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 从外部数据源获取一个 DataStream
DataStream<Tuple3<Long, String, Integer>> ds = env.addSource(...);
// 查询一个未注册的 Table
Table table = tableEnv.fromDataStream(ds, $("user"), $("product"), $("amount"));
Table result = tableEnv.sqlQuery(
"SELECT SUM(amount) FROM " + table + " WHERE product LIKE '%Rubber%'");
// 查询一个注册的 Table
tableEnv.createTemporaryView("Orders", ds, $("user"), $("product"), $("amount"));
// 执行 sqlQuery() 返回 Table 对象
Table result2 = tableEnv.sqlQuery(
"SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");
final Schema schema = new Schema()
.field("product", DataTypes.STRING())
.field("amount", DataTypes.INT());
// 创建并注册 TableSink
tableEnv.connect(new FileSystem().path("/path/to/file"))
.withFormat(...)
.withSchema(schema)
.createTemporaryTable("RubberOrders");
// 调用 executeSql() 执行 INSERT SQL,查询结果写入 TableSink
tableEnv.executeSql(
"INSERT INTO RubberOrders SELECT product, amount FROM Orders WHERE product LIKE '%Rubber%'");
执行查询
SELECT 语句或者 VALUES 语句可以通过 TableEnvironment.executeSql()
方法来执行,该方法返回 TableResult 对象用于包装查询的结果,一个 Table 对象可以通过 Table.execute()
方法执行获取查询结果。TableResult.collect()
方法返回一个可以关闭的行迭代器(除非所有的数据都被收集到本地,否则一个查询作业永远不会结束。所以通过 CloseableIterator#close()
方法主动地关闭作业以防止资源泄露)。 还可以通过 TableResult.print()
方法将查询结果打印到控制台。TableResult 中的结果数据只能被访问一次,因此一个 TableResult 实例中,collect()
方法和 print()
方法不能被同时使用。
TableResult.collect()
与 TableResult.print()
的行为在不同的 checkpointing 模式下略有不同。
- 对于批作业或没有配置任何 checkpointing 的流作业,
TableResult.collect()
与TableResult.print()
既不保证Exactly-once
、也不保证At-least-once
。查询结果在产生后可被客户端即刻访问,但作业失败和重启时将会报错。 - 对于配置了 Exactly-once checkpointing 的流作业,
TableResult.collect()
与TableResult.print()
保证Exactly-once
。一条结果数据只有在其关联的 checkpointing 完成后才能在客户端被访问。 - 对于配置了 At-least-once checkpointing 的流作业,
TableResult.collect()
与TableResult.print()
保证At-least-once
。查询结果在产生后可被客户端即刻访问,同一条结果可能被多次传递给客户端。
语法
Flink 通过支持标准 ANSI SQL的 Apache Calcite 解析 SQL。以下“BNF-语法”描述了批处理和流处理查询中所支持的 SQL 特性的超集。
query:
values
| {
select
| selectWithoutFrom
| query UNION [ ALL ] query
| query EXCEPT query
| query INTERSECT query
}
[ ORDER BY orderItem [, orderItem ]* ]
[ LIMIT { count | ALL } ]
[ OFFSET start { ROW | ROWS } ]
[ FETCH { FIRST | NEXT } [ count ] { ROW | ROWS } ONLY]
orderItem:
expression [ ASC | DESC ]
select:
SELECT [ ALL | DISTINCT ]
{ * | projectItem [, projectItem ]* }
FROM tableExpression
[ WHERE booleanExpression ]
[ GROUP BY { groupItem [, groupItem ]* } ]
[ HAVING booleanExpression ]
[ WINDOW windowName AS windowSpec [, windowName AS windowSpec ]* ]
selectWithoutFrom:
SELECT [ ALL | DISTINCT ]
{ * | projectItem [, projectItem ]* }
projectItem:
expression [ [ AS ] columnAlias ]
| tableAlias . *
tableExpression:
tableReference [, tableReference ]*
| tableExpression [ NATURAL ] [ LEFT | RIGHT | FULL ] JOIN tableExpression [ joinCondition ]
joinCondition:
ON booleanExpression
| USING '(' column [, column ]* ')'
tableReference:
tablePrimary
[ matchRecognize ]
[ [ AS ] alias [ '(' columnAlias [, columnAlias ]* ')' ] ]
tablePrimary:
[ TABLE ] tablePath [ dynamicTableOptions ] [systemTimePeriod] [[AS] correlationName]
| LATERAL TABLE '(' functionName '(' expression [, expression ]* ')' ')'
| UNNEST '(' expression ')'
tablePath:
[ [ catalogName . ] schemaName . ] tableName
systemTimePeriod:
FOR SYSTEM_TIME AS OF dateTimeExpression
dynamicTableOptions:
/*+ OPTIONS(key=val [, key=val]*) */
key:
stringLiteral
val:
stringLiteral
values:
VALUES expression [, expression ]*
groupItem:
expression
| '(' ')'
| '(' expression [, expression ]* ')'
| CUBE '(' expression [, expression ]* ')'
| ROLLUP '(' expression [, expression ]* ')'
| GROUPING SETS '(' groupItem [, groupItem ]* ')'
windowRef:
windowName
| windowSpec
windowSpec:
[ windowName ]
'('
[ ORDER BY orderItem [, orderItem ]* ]
[ PARTITION BY expression [, expression ]* ]
[
RANGE numericOrIntervalExpression {PRECEDING}
| ROWS numericExpression {PRECEDING}
]
')'
matchRecognize:
MATCH_RECOGNIZE '('
[ PARTITION BY expression [, expression ]* ]
[ ORDER BY orderItem [, orderItem ]* ]
[ MEASURES measureColumn [, measureColumn ]* ]
[ ONE ROW PER MATCH ]
[ AFTER MATCH
( SKIP TO NEXT ROW
| SKIP PAST LAST ROW
| SKIP TO FIRST variable
| SKIP TO LAST variable
| SKIP TO variable )
]
PATTERN '(' pattern ')'
[ WITHIN intervalLiteral ]
DEFINE variable AS condition [, variable AS condition ]*
')'
measureColumn:
expression AS alias
pattern:
patternTerm [ '|' patternTerm ]*
patternTerm:
patternFactor [ patternFactor ]*
patternFactor:
variable [ patternQuantifier ]
patternQuantifier:
'*'
| '*?'
| '+'
| '+?'
| '?'
| '??'
| '{' { [ minRepeat ], [ maxRepeat ] } '}' ['?']
| '{' repeat '}'
Flink SQL 对于标识符(表、属性、函数名)的命名策略类似于 Java 的词法约定:
-
标识符大小写敏感
-
通过反引号,可以允许标识符带有非字母的字符
SELECT a AS `my field` FROM t
字符串文本常量需要被单引号包起来(如 SELECT 'Hello World'
)。两个单引号表示转义(如 SELECT 'It''s me.'
)。字符串文本常量支持 Unicode 字符,如需明确使用 Unicode 编码,请使用以下语法:
- 使用反斜杠(\)作为转义字符(默认):
SELECT U&'\263A'
- 使用自定义的转义字符:
SELECT U&'#263A' UESCAPE '#'
操作符
WITH
WITH 提供了编写辅助语句的方法,以便在更大的查询中使用。这些语句通常被称为公共表表达式(Common Table Expression,CTE),可以认为它定义了只存在于一个查询中的临时视图。
WITH 语法:
WITH <with_item_definition> [ , ... ]
SELECT ... FROM ...;
<with_item_defintion>:
with_item_name (column_name[, ...n]) AS ( <select_query> )
下面的示例定义了一个 CTE:orders_with_total
,并在 GROUP BY 查询中使用它。
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;
SELECT & WHERE
SELECT 语句的一般语法为:
SELECT select_list FROM table_expression [ WHERE boolean_expression ]
table_expression
可以是任何数据源(表、视图、VALUES 子句、多个表的 Join 结果、子查询)。下面的事例读取 Orders 表的所有列:
SELECT * FROM Orders
select_list
指定 *
表示解析所有的列,但是不建议在生产环境中使用,会降低性能,建议只查询需要的列:
SELECT order_id, price + tax FROM Orders
查询可以使用 VALUES 子句,每个元组(Tuple)对应一个 Row,并且可以设置别名:
SELECT order_id, price FROM (VALUES (1, 2.0), (2, 3.1)) AS t (order_id, price)
WHERE 语句可以过滤 Row:
SELECT price + tax FROM Orders WHERE id = 10
可以对每行数据的指定列调用函数(内置、自定义函数,自定义函数必须提前注册):
SELECT PRETTY_PRINT(order_id) FROM Orders
SELECT DISTINCT
如果指定 SELECT DISTINCT,则将从结果集中删除重复行(每组重复中保留一行)。
SELECT DISTINCT id FROM Orders
对于流式查询,计算查询结果所需的状态(State)可能会无限增长。状态大小取决于不同行的数量。可<u>以为查询配置适当的状态生存时间(TTL),以防止状态大小过大。这可能会影响查询结果的正确性</u>。
Windowing TVF(1.13)
Window 是流处理的核心。Windows 将流拆分为有限大小的片段应用计算。只有流处理支持。
Flink 1.13 提供了几个 Table-valued functions(TVF,区别于 Group Window Function),将表中的元素划分为 windows,包括:
- 滚动窗口(Tumbling windows)
- 滑动窗口(Hop, Sliding windows)
- 累加窗口(Cumulate windows)
- 会话窗口(Session windows,TVF 暂不支持)
每个元素在逻辑上可以属于多个窗口,具体取决于所使用的窗口函数。TVF 必须和聚合操作一起使用:
假设存在一个 Bid
表
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 |
+------------------+-------+------+
滚动窗口(Tumbling windows)
指定一个固定大小的窗口,并且不重叠,语法:
TUMBLE(TABLE data, DESCRIPTOR(timecol), size)
-- data: 表名,表需要有时间属性字段
-- timecol: 表中的时间属性字段,用于划分窗口
-- size: 窗口大小
设定一个10分钟大小的滚动窗口,
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 |
+------------------+------------------+-------+
滑动窗口(Hop, Sliding windows)
指定一个固定大小的窗口,设定滑动间隔,元素会被指定给多个窗口,语法:
HOP(TABLE data, DESCRIPTOR(timecol), slide, size [, offset ])
-- data: 表名,表需要有时间属性字段
-- timecol: 表中的时间属性字段,用于划分窗口
-- size: 窗口大小
-- slide:窗口滑动的大小
设定一个10分钟大小,每5分钟滑动的窗口,
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 |
+------------------+------------------+-------+
累加窗口(Cumulate windows)
指定一个窗口的最大规模,按照指定时间间隔增长累加,直到达到窗口的最大规模,每次窗口增长会进行一次计算,可以理解为多次计算的滚动窗口,语法:
CUMULATE(TABLE data, DESCRIPTOR(timecol), step, size)
-- data: 表名,表需要有时间属性字段
-- timecol: 表中的时间属性字段,用于划分窗口
-- size: 窗口最大大小
-- step:窗口增长大小
设定一个10分钟大小,每2分钟累计一次的窗口,
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 |
+------------------+------------------+-------+
网友评论