Flink常见SQL语法说明

SELECT与WHERE语句

根据WHERE子句中的条件对数据进行过滤，仅返回满足条件的数据行。

• 语法：

  SELECT select_list FROM table_expression [ WHERE boolean_expression ]

• 示例：

  SELECT price + tax FROM Orders WHERE id = 10

WITH语句

WITH子句提供了一种用于更大查询而编写辅助语句的方法。这些语句通常被称为公用表表达式，表达式可以理解为仅针对某个查询而存在的临时视图。适用于复杂查询中将子查询抽象为可复用的临时表的场景。

该语句适用于MRS 3.1.2-LTS及以上版本集群中的Flink。

• 语法：

  WITH <with_item_definition> [ , ... ]

  SELECT ... FROM ...;

  <with_item_definition>:

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

• 示例：

  定义一个公用表表达式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;

窗口聚合语句

通过窗口聚合进行分组的查询将计算每个组的单个结果行。其中窗口通过TVF（Table-Valued Function，表值函数）定义，即将窗口作为一个函数返回一张表，包含window_start和window_end字段。适用于需要按时间窗口对数据进行聚合统计的场景。

该语句适用于MRS 3.1.2-LTS及以上版本集群中的Flink。

• 语法：

  SELECT ...

  FROM <windowed_table> -- relation applied windowing TVF

  GROUP BY window_start, window_end, …

• 示例：

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

窗口去重语句

Window Deduplication（窗口去重）是一种特殊的重复数据删除，它删除在一组列上重复的行，为每个窗口和分区键保留第一个或最后一个。适用于需要在时间窗口内去除重复数据的场景，例如每个窗口内只保留最新的一条记录。

该语句适用于MRS 3.2.0-LTS及以上版本集群中的Flink。

• 语法：

  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) -- relation applied windowing TVF

  WHERE (rownum = 1 | rownum <=1 | rownum < 2) [AND conditions]

• 示例：

  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' MINUTES))
  ) WHERE rownum <= 1;

Top-N语句

Top-N查询要求按列排序的N个最小值或最大值。适用于需要获取排名前N名的场景，例如查询销量前5的商品。

该语句适用于MRS 3.1.2-LTS及以上版本集群中的Flink。

• 语法：

  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]

• 示例：

  CREATE TABLE ShopSales (
  product_id   STRING,
  category     STRING,
  product_name STRING,
  sales        BIGINT
  ) WITH (...);
  
  SELECT *
  FROM (
  SELECT *,
  ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales DESC) AS row_num
  FROM ShopSales)
  WHERE row_num <= 5