flink---window

Window先容

DataStream:
https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/dev/datastream/operators/windows/
SQL:
https://nightlies.apache.org/flink/flink-docs-release-1.17/zh/docs/dev/table/sql/queries/window-tvf/
1、为什么必要Window? 

   在及时盘算范畴, 经常会有如下的需求:
  每隔xx时间, 盘算迩来xx时间的数据,
  如:
  每隔10min,盘算迩来24h的热搜词
  每隔5s,盘算迩来1min的股票行情数据
  每隔10min,盘算迩来1h的广告点击量
  ....
  这些及时需求的实现就必要借助窗口!
  2、Window有哪些控制属性? 

   为了完成上面提到的需求, 必要使用窗口来完成, 但是窗口必要有如下的属性才可以
  窗口的长度(巨细): 决定了要盘算迩来多长时间的数据
  窗口的隔断: 决定了每隔多久盘算一次
   3、基于时间的滑动和滚动窗口

窗口的长度(巨细) > 窗口的隔断 : 如每隔5s, 盘算迩来10s的数据 【滑动窗口】
 

窗口的长度(巨细) = 窗口的隔断: 如每隔10s,盘算迩来10s的数据 【滚动窗口】

窗口的长度(巨细) < 窗口的隔断: 每隔15s,盘算迩来10s的数据 【没闻名字,不消】

滚动窗口 Tumble （DataStream Tumbling Window）
滑动窗口 HOP （DataStream Sliding Window）
累积窗口 Cumulate (DataStream没有)
     在实际应用中还会遇到如许一类需求：我们的统计周期大概较长，因此盼望中央每隔一段时间就输出一次当前的统计值；与滑动窗口差别的是，在一个统计周期内，我们会多次输出统计值，它们应该是不停叠加累积的。这种特殊的窗口就叫作“累积窗口”（Cumulate Window），它会在肯定的统计周期内举行累积盘算。累积窗口中有两个核心的参数：最大窗口长度（max window size）和累积步长（step）。所谓的最大窗口长度着实就是我们所说的“统计周期”，终极目的就是统计这段时间内的数据。开始时，创建的第一个窗口巨细就是步长 step；之后的每个窗口都会在之前的底子上再扩展 step 的长度，直到到达最大窗口长度。在 SQL 中可以用 CUMULATE()函数来界说，具体如下：

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

复制代码

        这里我们基于时间属性 ts，在表 EventTable 上界说了一个统计周期为 1 天、累积步长为 1 小时的累积窗口。注意第三个参数为步长 step ，第四个参数则是最大窗口长度。

4、processTime Window 

1. /**
2. *
3. * {"username":"zs","price":20}
4. * {"username":"lisi","price":15}
5. * {"username":"lisi","price":20}
6. * {"username":"zs","price":20}
7. * {"username":"zs","price":20}
8. * {"username":"zs","price":20}
9. * {"username":"zs","price":20}
10. *
11. */
12. //窗口触发的条件  1. 系统时间大于等于窗口的结束时间   2. 窗口内有数据
13. //滚动窗口 TUMBLE(TABLE KafkaTable, DESCRIPTOR(event_time), INTERVAL '10' SECOND)
14. //滑动窗口 :每隔10秒，计算最近10秒数据。统计每个用户在最近10秒消费的次数和总金额
15. CREATE TABLE KafkaTable (
16.   `username`    STRING,
17.   `price`       INT,
18.   `event_time`  as proctime()     -- 计算列
19. ) WITH (
20.   'connector' = 'kafka',
21.   'topic' = 'topic1',
22.   'properties.bootstrap.servers' = 'hadoop11:9092,hadoop12:9092,hadoop13:9092',
23.   'properties.group.id' = 'testGroup',
24.   'scan.startup.mode' = 'latest-offset',
25.   'format' = 'json'
26. );
27. select username,window_start,window_end,count(*) cnt,sum(price) total_price
28. from table(TUMBLE(TABLE KafkaTable, DESCRIPTOR(event_time), INTERVAL '10' SECOND))
29. group by username,window_start,window_end;
31. //滑动窗口：每隔30秒，计算最近1分钟每隔用户消费次数和消费总金额。
32. CREATE TABLE KafkaTable (
33.   `username`    STRING,
34.   `price`       INT,
35.   `event_time`  as proctime()     -- 计算列
36. ) WITH (
37.   'connector' = 'kafka',
38.   'topic' = 'topic1',
39.   'properties.bootstrap.servers' = 'hadoop11:9092,hadoop12:9092,hadoop13:9092',
40.   'properties.group.id' = 'testGroup',
41.   'scan.startup.mode' = 'latest-offset',
42.   'format' = 'json'
43. );
44. select username,window_start,window_end,count(*) cnt,sum(price) total_price
45. from table(HOP(TABLE KafkaTable, DESCRIPTOR(event_time), INTERVAL '30' SECOND, INTERVAL '60' SECOND))
46. group by username,window_start,window_end;

复制代码