Flink SQL

Data Types

链接

支持结构化类型
ARRAY<t>
MAP<kt, vt>
MULTISET<t>
ROW<n0 t0, n1 t1, ...>

User-Defined Data Types

Aggregations

1. GroupBy Aggregation

//  GroupBy on a streaming table produces an updating result (持续更新的结果)
SELECT a, SUM(b) as d
FROM Orders
GROUP BY a

2. GroupBy Window Aggregation

// 使用分组窗口计算每个组的 single result（单个结果）
SELECT user, SUM(amount)
FROM Orders
GROUP BY TUMBLE(rowtime, INTERVAL '1' DAY), user

3. Over Window aggregation

OVER WINDOW 窗口是一种常用于计算数据流中的行与行之间的关系的窗口。与 GROUP BY 聚合不同，OVER 窗口不会将输入的行分组到有限的一组输出行中，而是为输入数据流中的每一行都进行聚合操作，产生一个结果行。所有聚合都必须在同一个窗口上定义，即相同的分区、排序和范围。

OVER WINDOW 的定义通常包括以下几部分：
PARTITION BY：这部分定义了如何对数据进行分组。在每个分组中，OVER WINDOW 会独立地进行计算。
ORDER BY：这部分定义了数据在每个分组中的排序方式。OVER WINDOW 的计算会按照这个顺序进行。ORDER BY 只能在单个时间属性上。
ROWS BETWEEN 或 RANGE BETWEEN：这部分定义了窗口的范围。例如，ROWS BETWEEN 2 PRECEDING AND CURRENT ROW 表示窗口范围是当前行及其前面的两行。
*/

/* 
 * PARTITION BY user 表示按照用户进行分组，ORDER BY proctime 表示按照处理时间进行排序
 * ROWS BETWEEN 2 PRECEDING AND CURRENT ROW 表示窗口范围是当前行及其前面的两行
 */
SELECT COUNT(amount) OVER w, SUM(amount) OVER w
FROM Orders
WINDOW w AS (
  PARTITION BY user
  ORDER BY proctime
  ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)

// 计算每个用户的滚动平均购买金额
SELECT user, AVG(amount) OVER (PARTITION BY user ORDER BY proctime ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)
FROM Orders

4. GROUPING SETS、ROLLUP 和 CUBE
GROUPING SETS：它允许你在一个查询中定义多个分组。例如，你可能想要按照城市和国家进行分组，也可能只想按照国家进行分组。GROUPING SETS 可以让你在一个查询中同时做到这两点。

// 这个查询将会返回两组结果：
// 按照国家和城市分组的销售总额。这意味着，对于每一个国家和城市的组合，都会有一个销售总额的结果。
// 按照国家分组的销售总额。这意味着，对于每一个国家，都会有一个销售总额的结果
// 这两组结果都会在同一个结果集中返回，你可以通过结果中的 NULL 值来区分这两组结果。
// 在第一组结果中，country 和 city 列都不会是 NULL；
// 在第二组结果中，country 列不会是 NULL，但是 city 列会是 NULL。
SELECT country, city, SUM(sales)
FROM sales
GROUP BY GROUPING SETS ((country, city), (country))

Joins

Inner Equi-join

对于流式查询，计算查询结果所需的状态可能会根据不同输入行的数量而无限增长。请提供具有有效保留间隔的查询配置，以防止状态大小过大

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

// Outer Equi-join
SELECT *
FROM Orders LEFT JOIN Product ON Orders.productId = Product.id

SELECT *
FROM Orders RIGHT JOIN Product ON Orders.productId = Product.id

SELECT *
FROM Orders FULL OUTER JOIN Product ON Orders.productId = Product.id

Inner/Outer Interval Join

间隔联接可以以流方式处理的常规联接的子集
间隔联接至少需要一个等联接谓词和一个限定两边时间的联接条件

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

SELECT *
FROM Orders o, Shipments s
WHERE o.id = s.orderId AND
      o.ordertime BETWEEN s.shiptime - INTERVAL '4' HOUR AND s.shiptime

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]

Deduplication

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

Group Windows

  SELECT user,
    TUMBLE_START(rowtime, INTERVAL '1' DAY) as wStart,
    SUM(amount) FROM Orders
   GROUP BY TUMBLE(rowtime, INTERVAL '1' DAY), user

 SELECT product, SUM(amount) FROM Orders GROUP BY HOP(rowtime, INTERVAL '1' HOUR, INTERVAL '1' DAY), product