Fork me on GitHub

Flink 原理与实现: FlinkSQL的Table API 与SQL之窗口（Windows）

字数统计: 2k字   |   阅读时长≈ 8分

​ 时间语义，要配合窗口操作才能发挥作用。最主要的用途，当然就是开窗口、根据时间段做计算了。下面我们就来看看Table API和SQL中，怎么利用时间字段做窗口操作。

在Table API和SQL中，主要有两种窗口：Group Windows和Over Windows

分组窗口（Group Windows）

分组窗口（Group Windows）会根据时间或行计数间隔，将行聚合到有限的组（Group）中，并对每个组的数据执行一次聚合函数。

Table API中的Group Windows都是使用.window（w:GroupWindow）子句定义的，并且必须由as子句指定一个别名。为了按窗口对表进行分组，窗口的别名必须在group by子句中，像常规的分组字段一样引用。

1
2
3
4
Table table = input
  .window([w: GroupWindow] as "w") // 定义窗口，别名 w
  .groupBy("w, a")  // 以属性a和窗口w作为分组的key 
  .select("a, b.sum")  // 聚合字段b的值，求和

或者，还可以把窗口的相关信息，作为字段添加到结果表中：

1
2
3
4
Table table = input
  .window([w: GroupWindow] as "w") 
  .groupBy("w, a") 
  .select("a, w.start, w.end, w.rowtime, b.count")

Table API提供了一组具有特定语义的预定义Window类，这些类会被转换为底层DataStream或DataSet的窗口操作。

Table API支持的窗口定义，和我们熟悉的一样，主要也是三种：滚动（Tumbling）、滑动（Sliding）和会话（Session）。

滚动窗口

滚动窗口（Tumbling windows）要用Tumble类来定义，另外还有三个方法：

  • over：定义窗口长度
  • on：用来分组（按时间间隔）或者排序（按行数）的时间字段
  • as：别名，必须出现在后面的groupBy中

代码如下：

1
2
3
4
5
6
7
8
// Tumbling Event-time Window
.window(Tumble.over("10.minutes").on("rowtime").as("w"))

// Tumbling Processing-time Window
.window(Tumble.over("10.minutes").on("proctime").as("w"))

// Tumbling Row-count Window
.window(Tumble.over("10.rows").on("proctime").as("w"))

滑动窗口

滑动窗口（Sliding windows）要用Slide类来定义，另外还有四个方法：

  • over：定义窗口长度
  • every：定义滑动步长
  • on：用来分组（按时间间隔）或者排序（按行数）的时间字段
  • as：别名，必须出现在后面的groupBy中

代码如下：

1
2
3
4
5
6
7
8
// Sliding Event-time Window
.window(Slide.over("10.minutes").every("5.minutes").on("rowtime").as("w"))

// Sliding Processing-time window 
.window(Slide.over("10.minutes").every("5.minutes").on("proctime").as("w"))

// Sliding Row-count window
.window(Slide.over("10.rows").every("5.rows").on("proctime").as("w"))

会话窗口

会话窗口（Session windows）要用Session类来定义，另外还有三个方法：

  • withGap：会话时间间隔
  • on：用来分组（按时间间隔）或者排序（按行数）的时间字段
  • as：别名，必须出现在后面的groupBy中

代码如下：

1
2
3
4
5
// Session Event-time Window
.window(Session.withGap.("10.minutes").on("rowtime").as("w"))

// Session Processing-time Window
.window(Session.withGap.("10.minutes").on(“proctime").as("w"))

Over Windows

Over window聚合是标准SQL中已有的（Over子句），可以在查询的SELECT子句中定义。Over window 聚合，会针对每个输入行，计算相邻行范围内的聚合。Over windows

使用.window（w:overwindows*）子句定义，并在select（）方法中通过别名来引用。

比如这样：

1
2
3
Table table = input
  .window([w: OverWindow] as "w")
  .select("a, b.sum over w, c.min over w")

Table API提供了Over类，来配置Over窗口的属性。可以在事件时间或处理时间，以及指定为时间间隔、或行计数的范围内，定义Over windows。

无界的over window是使用常量指定的。也就是说，时间间隔要指定UNBOUNDED_RANGE，或者行计数间隔要指定UNBOUNDED_ROW。而有界的over window是用间隔的大小指定的。

实际代码应用如下：

1） 无界的 over window

1
2
3
4
5
6
7
8
9
10
11
// 无界的事件时间 over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding.(UNBOUNDED_RANGE).as("w"))

// 无界的处理时间 over window
.window(Over.partitionBy("a").orderBy("proctime").preceding.(UNBOUNDED_RANGE).as("w"))

// 无界的事件时间 Row-count over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding.(UNBOUNDED_ROW).as("w"))

//无界的处理时间 Row-count over window
.window(Over.partitionBy("a").orderBy("proctime").preceding.(UNBOUNDED_ROW).as("w"))

2） 有界的over window

1
2
3
4
5
6
7
8
9
10
11
// 有界的事件时间 over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding("1.minutes").as("w"))
        
// 有界的处理时间 over window
.window(Over.partitionBy("a").orderBy("proctime").preceding("1.minutes").as("w"))
        
// 有界的事件时间 Row-count over window
.window(Over.partitionBy("a").orderBy("rowtime").preceding("10.rows").as("w"))
        
// 有界的处理时间 Row-count over window
.window(Over.partitionBy("a").orderBy("procime").preceding("10.rows").as("w"))

SQL中窗口的定义

我们已经了解了在Table API里window的调用方式，同样，我们也可以在SQL中直接加入窗口的定义和使用。

Group Windows

Group Windows在SQL查询的Group BY子句中定义。与使用常规GROUP BY子句的查询一样，使用GROUP BY子句的查询会计算每个组的单个结果行。

SQL支持以下Group窗口函数:

  1. TUMBLE(time_attr, interval)

​ 定义一个滚动窗口，第一个参数是时间字段，第二个参数是窗口长度。

  1. HOP(time_attr, interval, interval)

​ 定义一个滑动窗口，第一个参数是时间字段，第二个参数是窗口滑动步长，第三个是窗口长度。

  1. SESSION(time_attr, interval)

​ 定义一个会话窗口，第一个参数是时间字段，第二个参数是窗口间隔（Gap）。

另外还有一些辅助函数，可以用来选择Group Window的开始和结束时间戳，以及时间属性。

这里只写TUMBLE_*，滑动和会话窗口是类似的（HOP_*，SESSION_*）。

  • TUMBLE_START(time_attr, interval)
  • TUMBLE_END(time_attr, interval)
  • TUMBLE_ROWTIME(time_attr, interval)
  • TUMBLE_PROCTIME(time_attr, interval)

Over Windows

​ 由于Over本来就是SQL内置支持的语法，所以这在SQL中属于基本的聚合操作。所有聚合必须在同一窗口上定义，也就是说，必须是相同的分区、排序和范围。目前仅支持在当前行范围之前的窗口（无边界和有边界）。

注意，ORDER BY必须在单一的时间属性上指定。

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
SELECT COUNT(amount) OVER (
  PARTITION BY user
  ORDER BY proctime
  ROWS BETWEEN 2 PRECEDING AND CURRENT ROW)
FROM Orders

// 也可以做多个聚合
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)

代码练习（以分组滚动窗口为例）

我们可以综合学习过的内容，用一段完整的代码实现一个具体的需求。例如，可以开一个滚动窗口，统计10秒内出现的每个sensor的个数。

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
public static void main(String[] args) throws Exception {
        // 1. 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 设置事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2. 读取文件，得到 DataStream
        String filePath = "..\\sensor.txt";
        DataStream<String> inputStream = env.readTextFile(filePath);

        // 3. 转换成 Java Bean，并指定timestamp和watermark
        DataStream<SensorReading> dataStream = inputStream
                .map( line -> {
                    String[] fields = line.split(",");
                    return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
                } )
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(1)) {
                    @Override
                    public long extractTimestamp(SensorReading element) {
                        return element.getTimestamp() * 1000L;
                    }
                });

        // 4. 将 DataStream 转换为 Table
        Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp.rowtime as ts, temperature");

        // 5. 开窗聚合
        Table resultTable = sensorTable
                .window(Tumble.over("10.seconds").on("ts").as("tw"))
                // 每10秒统计一次，滚动时间窗口
                .groupBy("id, tw")
                .select("id, id.count, temperature.avg, tw.end");

        //  sql
        tableEnv.createTemporaryView("sensor", sensorTable);
        Table resultSqlTable = tableEnv.sqlQuery(
                "select id, count(id), avg(temperature), tumble_end(ts, interval '10' second) " +
                        "from sensor group by id, tumble(ts, interval '10' second)");

        // 转换成流打印输出
        tableEnv.toAppendStream(resultTable, Row.class).print("result");
        tableEnv.toAppendStream(resultSqlTable, Row.class).print("sql");

        env.execute("time and window test");
    }
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

标签列表:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
大数据开发、Java后端开发
联系QQ:3101001329
微信:shangbaishuyao

很惭愧

只做了一点微小的工作
谢谢大家