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

最后编辑于：2023.10.20 18:45:20