（2）滑动窗口（HOP）

滑动窗口的使用与滚动窗口类似，可以通过设置滑动步长来控制统计输出的频率。在SQL中通过调用HOP()来声明滑动窗口；除了也要传入表名、时间属性外，还需要传入窗口大小（size）和滑动步长（slide）两个参数。

HOP(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '5' MINUTES, INTERVAL '1' HOURS));

需要注意的是，紧跟在时间属性字段后面的第三个参数是步长（slide），第四个参数才是窗口大小（size）。

（3）累积窗口（CUMULATE）

累积窗口是窗口TVF中新增的窗口功能，它会在一定的统计周期内进行累积计算。累积窗口中有两个核心的参数：最大窗口长度（max window size）和累积步长（step）。所谓的最大窗口长度其实就是我们所说的“统计周期”，最终目的就是统计这段时间内的数据在SQL中可以用CUMULATE()函数来定义，具体如下：

CUMULATE(TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOURS, INTERVAL '1' DAYS))

注意第三个参数为步长step，第四个参数则是最大窗口长度。上面所有的语句只是定义了窗口，类似于DataStream API中的窗口分配器；在SQL中窗口的完整调用，还需要配合聚合操作和其它操作。

五、聚合（Aggregation）查询

Flink 中的SQL是流处理与标准SQL结合的产物，所以聚合查询也可以分成两种：流处理中特有的聚合（主要指窗口聚合），以及SQL原生的聚合查询方式。

5.1 分组聚合

SQL中一般所说的聚合我们都很熟悉，主要是通过内置的一些聚合函数来实现的，比如SUM()、MAX()、MIN()、AVG()以及COUNT()。它们的特点是对多条输入数据进行计算，得到一个唯一的值，属于“多对一”的转换。比如我们可以通过下面的代码计算输入数据的个数：

Table eventCountTable = tableEnv.sqlQuery("select COUNT(*) from EventTable");

而更多的情况下，我们可以通过GROUP BY子句来指定分组的键（key），从而对数据按照某个字段做一个分组统计。例如之前我们举的例子，可以按照用户名进行分组，统计每个用户点击url的次数：

SELECT user, COUNT(url) as cnt FROM EventTable GROUP BY user

这种聚合方式，就叫作“分组聚合”（group aggregation）。想要将结果表转换成流或输出到外部系统，必须采用撤回流（retract stream）或更新插入流（upsert stream）的编码方式；如果在代码中直接转换成DataStream打印输出，需要调用toChangelogStream()。

分组聚合既是SQL原生的聚合查询，也是流处理中的聚合操作，这是实际应用中最常见的聚合方式。当然，使用的聚合函数一般都是系统内置的，如果希望实现特殊需求也可以进行自定义。

5.2 窗口聚合

在Flink的Table API和SQL中，窗口的计算是通过“窗口聚合”（window aggregation）来实现的。与分组聚合类似，窗口聚合也需要调用SUM()、MAX()、MIN()、COUNT()一类的聚合函数，通过GROUP BY子句来指定分组的字段。只不过窗口聚合时，需要将窗口信息作为分组key的一部分定义出来。

在Flink 1.12版本之前，是直接把窗口自身作为分组key放在GROUP BY之后的，所以也叫“分组窗口聚合”；而1.13版本开始使用了“窗口表值函数”（Windowing TVF），窗口本身返回的是就是一个表，所以窗口会出现在FROM后面，GROUP BY后面的则是窗口新增的字段window_start和window_end。例如：

Table result = tableEnv.sqlQuery(
                        "SELECT " +
                            "user, " +
                            "window_end AS endT, " +
                            "COUNT(url) AS cnt " +
                        "FROM TABLE( " +
                                  "TUMBLE( TABLE EventTable, " +
                                  "DESCRIPTOR(ts), " +
                                  "INTERVAL '1' HOUR)) " +
                        "GROUP BY user, window_start, window_end "
                );

Flink SQL目前提供了滚动窗口TUMBLE()、滑动窗口HOP()和累积窗口（CUMULATE）三种表值函数（TVF）。在具体应用中，我们还需要提前定义好时间属性。下面是一段窗口聚合的完整代码，以累积窗口为例：

public class CumulateWindowExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
// 读取数据源，并分配时间戳、生成水位线
        SingleOutputStreamOperator<Event> eventStream = env
                .fromElements(
                        new Event("Alice", "./home", 1000L),
                        new Event("Bob", "./cart", 1000L),
                        new Event("Alice", "./prod?id=1",  25 * 60 * 1000L),
                        new Event("Alice", "./prod?id=4", 55 * 60 * 1000L),
                        new Event("Bob", "./prod?id=5", 3600 * 1000L + 60 * 1000L),
                        new Event("Cary", "./home", 3600 * 1000L + 30 * 60 * 1000L),
                        new Event("Cary", "./prod?id=7", 3600 * 1000L + 59 * 60 * 1000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );
// 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
// 将数据流转换成表，并指定时间属性
        Table eventTable = tableEnv.fromDataStream(
                eventStream,
                $("user"),
                $("url"),
                $("timestamp").rowtime().as("ts")  
        );
// 为方便在SQL中引用，在环境中注册表EventTable
        tableEnv.createTemporaryView("EventTable", eventTable);
// 设置累积窗口，执行SQL统计查询
        Table result = tableEnv
                .sqlQuery(
                        "SELECT " +
                            "user, " +
                            "window_end AS endT, " +
                            "COUNT(url) AS cnt " +
                        "FROM TABLE( " +
                            "CUMULATE( TABLE EventTable, " +    // 定义累积窗口
                                "DESCRIPTOR(ts), " +
                                "INTERVAL '30' MINUTE, " +
                                "INTERVAL '1' HOUR)) " +
                        "GROUP BY user, window_start, window_end "
                );
        tableEnv.toDataStream(result).print();
        env.execute();
    }
}

基于窗口的聚合，是流处理中聚合统计的一个特色，也是与标准SQL最大的不同之处。在实际项目中，很多统计指标其实都是基于时间窗口来进行计算的，所以窗口聚合是Flink SQL中非常重要的功能；基于窗口TVF的聚合未来也会有更多功能的扩展支持，比如窗口TOP-N、会话窗口、窗口联结等等。

5.3 开窗（Over）聚合

在标准SQL中还有另外一类比较特殊的聚合方式，可以针对每一行计算一个聚合值，这就是所谓的“开窗函数”。开窗函数的聚合与之前两种聚合有本质的不同：分组聚合、窗口TVF聚合都是“多对一”的关系，将数据分组之后每组只会得到一个聚合结果；而开窗函数是对每行都要做一次开窗聚合，因此聚合之后表中的行数不会有任何减少，是一个“多对多”的关系。

与标准SQL中一致，Flink SQL中的开窗函数也是通过OVER子句来实现的，所以有时开窗聚合也叫作“OVER聚合”（Over Aggregation）。基本语法如下：

SELECT
  <聚合函数> OVER (
    [PARTITION BY <字段1>[, <字段2>, ...]]
    ORDER BY <时间属性字段>
    <开窗范围>),
  ...
FROM ...

这里OVER关键字前面是一个聚合函数，它会应用在后面OVER定义的窗口上。在OVER子句中主要有以下几个部分：

PARTITION BY（可选）

用来指定分区的键（key），类似于GROUP BY的分组，这部分是可选的；

ORDER BY

在OVER子句中必须用ORDER BY明确地指出数据基于那个字段排序。在Flink的流处理中，目前只支持按照时间属性的升序排列，所以这里ORDER BY后面的字段必须是定义好的时间属性。

开窗范围

由BETWEEN <下界> AND <上界> 定义，也就是“从下界到上界”的范围。目前支持的上界只能是CURRENT ROW，也就是定义一个“从之前某一行到当前行”的范围，所以一般的形式为：

BETWEEN ... PRECEDING AND CURRENT ROW

开窗选择的范围可以基于时间，也可以基于数据的数量。所以开窗范围还应该在两种模式之间做出选择：范围间隔（RANGE intervals）和行间隔（ROW intervals）。

范围间隔

范围间隔以RANGE为前缀，就是基于ORDER BY指定的时间字段去选取一个范围，一般就是当前行时间戳之前的一段时间。例如开窗范围选择当前行之前1小时的数据：

RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW

行间隔

行间隔以ROWS为前缀，就是直接确定要选多少行，由当前行出发向前选取就可以了。例如开窗范围选择当前行之前的5行数据（最终聚合会包括当前行，所以一共6条数据）：

ROWS BETWEEN 5 PRECEDING AND CURRENT ROW

下面是一个具体示例：

SELECT user, ts,
        COUNT(url) OVER (
            PARTITION BY user
            ORDER BY ts
            RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW
        ) AS cnt
FROM EventTable

开窗聚合与窗口聚合（窗口TVF聚合）本质上不同，不过也还是有一些相似之处的：它们都是在无界的数据流上划定了一个范围，截取出有限数据集进行聚合统计；这其实都是“窗口”的思路。事实上，在Table API中确实就定义了两类窗口：分组窗口（GroupWindow）和开窗窗口（OverWindow）；而在SQL中，也可以用WINDOW子句来在SELECT外部单独定义一个OVER窗口：

SELECT user, ts,
  COUNT(url) OVER w AS cnt,
  MAX(CHAR_LENGTH(url)) OVER w AS max_url
FROM EventTable
WINDOW w AS (
  PARTITION BY user
  ORDER BY ts
  ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)

上面的SQL中定义了一个选取之前2行数据的OVER窗口，并重命名为w；接下来就可以基于它调用多个聚合函数，扩展出更多的列提取出来。

5.4 应用实例 —— TOP-N

目前在Flink SQL中没有能够直接调用的TOP-N函数，而是提供了稍微复杂些的变通实现方法。下面是一个具体案例的代码实现。由于用户访问事件Event中没有商品相关信息，因此我们统计每小时内有最多访问行为的用户，取前两名，相当于是一个每小时活跃用户的查询。

public class WindowTopNExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
// 读取数据源，并分配时间戳、生成水位线
        SingleOutputStreamOperator<Event> eventStream = env
                .fromElements(
                        new Event("Alice", "./home", 1000L),
                        new Event("Bob", "./cart", 1000L),
                        new Event("Alice", "./prod?id=1",  25 * 60 * 1000L),
                        new Event("Alice", "./prod?id=4", 55 * 60 * 1000L),
                        new Event("Bob", "./prod?id=5", 3600 * 1000L + 60 * 1000L),
                        new Event("Cary", "./home", 3600 * 1000L + 30 * 60 * 1000L),
                        new Event("Cary", "./prod?id=7", 3600 * 1000L + 59 * 60 * 1000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );
        // 创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 将数据流转换成表，并指定时间属性
        Table eventTable = tableEnv.fromDataStream(
                eventStream,
                $("user"),
                $("url"),
                $("timestamp").rowtime().as("ts")  
// 将timestamp指定为事件时间，并命名为ts
        );
        // 为方便在SQL中引用，在环境中注册表EventTable
        tableEnv.createTemporaryView("EventTable", eventTable);
        // 定义子查询，进行窗口聚合，得到包含窗口信息、用户以及访问次数的结果表
        String subQuery =
                "SELECT window_start, window_end, user, COUNT(url) as cnt " +
                "FROM TABLE ( " +
                    "TUMBLE( TABLE EventTable, DESCRIPTOR(ts), INTERVAL '1' HOUR )) " +
                "GROUP BY window_start, window_end, user ";
        // 定义TOP-N的外层查询
        String topNQuery =
                "SELECT * " +
                "FROM (" +
                    "SELECT *, " +
                        "ROW_NUMBER() OVER ( " +
                            "PARTITION BY window_start, window_end " +
                            "ORDER BY cnt desc " +
                        ") AS row_num " +
                    "FROM (" + subQuery + ")) " +
                "WHERE row_num <= 2";
        // 执行SQL得到结果表
        Table result = tableEnv.sqlQuery(topNQuery);
        tableEnv.toDataStream(result).print();
        env.execute();
    }
}

六、联结（Join）查询

在标准SQL中，可以将多个表连接合并起来，从中查询出想要的信息；这种操作就是表的联结（Join）。在Flink SQL中，同样支持各种灵活的联结（Join）查询，操作的对象是动态表。在流处理中，动态表的Join对应着两条数据流的Join操作。Flink SQL中的联结查询大体上也可以分为两类：SQL原生的联结查询方式，和流处理中特有的联结查询。

6.1 常规联结查询

常规联结（Regular Join）是SQL中原生定义的Join方式，是最通用的一类联结操作。它的具体语法与标准SQL的联结完全相同，通过关键字JOIN来联结两个表，后面用关键字ON来指明联结条件。与标准SQL一致，Flink SQL的常规联结也可以分为内联结（INNER JOIN）和外联结（OUTER JOIN），区别在于结果中是否包含不符合联结条件的行。目前仅支持“等值条件”作为联结条件，也就是关键字ON后面必须是判断两表中字段相等的逻辑表达式。

1. 等值内联结（INNER Equi-JOIN）

内联结用INNER JOIN来定义，会返回两表中符合联接条件的所有行的组合，也就是所谓的笛卡尔积（Cartesian product）。目前仅支持等值联结条件。例如：

SELECT *
FROM Order
INNER JOIN Product
ON Order.product_id = Product.id

2. 等值外联结（OUTER Equi-JOIN）

与内联结类似，外联结也会返回符合联结条件的所有行的笛卡尔积；另外，还可以将某一侧表中找不到任何匹配的行也单独返回。Flink SQL支持左外（LEFT JOIN）、右外（RIGHT JOIN）和全外（FULL OUTER JOIN），分别表示会将左侧表、右侧表以及双侧表中没有任何匹配的行返回。具体用法如下：

SELECT *
FROM Order
LEFT JOIN Product
ON Order.product_id = Product.id
SELECT *
FROM Order
RIGHT JOIN Product
ON Order.product_id = Product.id
SELECT *
FROM Order
FULL OUTER JOIN Product
ON Order.product_id = Product.id

这部分知识与标准SQL中是完全一样的。

6.2 间隔联结查询

我们曾经学习过DataStream API中的双流Join，包括窗口联结（window join）和间隔联结（interval join）。两条流的Join就对应着SQL中两个表的Join，这是流处理中特有的联结方式。目前Flink SQL还不支持窗口联结，而间隔联结则已经实现。间隔联结（Interval Join）返回的，同样是符合约束条件的两条中数据的笛卡尔积。只不过这里的“约束条件”除了常规的联结条件外，还多了一个时间间隔的限制。具体语法有以下要点：

两表的联结

间隔联结不需要用JOIN关键字，直接在FROM后将要联结的两表列出来就可以，用逗号分隔。这与标准SQL中的语法一致，表示一个“交叉联结”（Cross Join），会返回两表中所有行的笛卡尔积。

联结条件

联结条件用WHERE子句来定义，用一个等值表达式描述。交叉联结之后再用WHERE进行条件筛选，效果跟内联结INNER JOIN ... ON ...非常类似。

时间间隔限制

我们可以在WHERE子句中，联结条件后用AND追加一个时间间隔的限制条件；做法是提取左右两侧表中的时间字段，然后用一个表达式来指明两者需要满足的间隔限制。具体定义方式有下面三种，这里分别用ltime和rtime表示左右表中的时间字段：

（1）ltime = rtime
（2）ltime >= rtime AND ltime < rtime + INTERVAL '10' MINUTE
（3）ltime BETWEEN rtime - INTERVAL '10' SECOND AND rtime + INTERVAL '5' SECOND

例如，我们现在除了订单表Order外，还有一个“发货表”Shipment，要求在收到订单后四个小时内发货。那么我们就可以用一个间隔联结查询，把所有订单与它对应的发货信息连接合并在一起返回。

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

Flink1.13架构全集| 一文带你由浅入深精通Flink方方面面（三）SQL篇（三）

五、聚合（Aggregation）查询

5.1 分组聚合

5.2 窗口聚合

5.3 开窗（Over）聚合

5.4 应用实例 —— TOP-N

六、联结（Join）查询

6.1 常规联结查询

6.2 间隔联结查询

热门文章

最新文章

相关课程

相关电子书

相关实验场景