FlinkSql的窗口使用以及运用案例

1 flinkSQL窗口概述

1.1 窗口定义:

可理解为时间轴，可将无界流切分成有界流

1.2 窗口分类：

1. TimeWindow：通过时间切割窗口，但是不知道窗口有多少数据

• 滑动窗口
• 滚动窗口
• 会话窗口

1. CountWindow：按照数据量来切割窗口

• 滑动窗口
• 滚动窗口
• 会话窗口

1. 自定义窗口

1.3 TimeWindow分类

• 滚动窗口：有固定的窗口长度往前进行滚动，数据不重复计算
• 滑动窗口：由固定的窗口长度和滑动间隔组成，数据可以重复

• 会话窗口：由一系列事件指定事件长度间隙组成，类比wed应用的session
• group windows- 键控window：keyvalue- 非键控window

2 flinkSQL窗口使用

2.1 窗口函数类型

flinkSQL中通过Groupby Windows函数来定义分组窗口

• TUMBLE(time_attr,interval)：定义滚动窗口
• HOP(time_attr,interval,interval)：定义滑动窗口，第二个参数表示滑动步长，第三个参数表示窗口大小
• SESSION(time_attr,interval)：定义会话窗口

2.2 滚动窗口案例

1. 数据

data_time,price,product_id,buyername
1666620609,44,1,白天磊
1666620610,45,1,陈智渊
1666620611,46,1,崔钰轩
1666620612,47,1,吴鹏飞
1666620613,48,1,毛明辉
1666620614,49,1,侯弘文
1666620615,50,1,曾伟祺
1666620616,51,1,郝瑞霖
1666620617,52,1,陆熠彤
1666620618,53,1,余弘文
1666620619,54,1,石哲瀚
1666620620,55,1,任擎苍
1666620621,56,1,卢文轩
1666620622,57,1,吕晋鹏
1666620623,58,1,罗晟睿
1666620624,59,1,周建辉
1666620625,60,1,卢皓轩
1666620626,61,1,沈煜城
1666620627,62,1,万鑫鹏
1666620628,63,1,沈思远

1. 需求

• 上表是product_id为1的商品被不同的用户在不同的时间下单以及金额数据，使用flinkSQL当中当中的滚动窗口计算：每隔2秒钟的金额的最大值

1. 代码实现

• 定义Userproduct类定义字段

//使用插件生成有无参构造器以及重写一些方法@Data//完成了Getter,Setter,equals,hasCode,toString 等方法@Builder//省去写很多构造函数的麻烦@NoArgsConstructor//自动添加一个无参构造函数@AllArgsConstructor//为自动添加一个构造函数publicclassUserproduct{privateInteger product_id;privateString buyer_name;privateLong date_time;privateDouble price;}

• 构造执行环境

StreamExecutionEnvironment senv =StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);//设置并行度StreamTableEnvironment tEnv =StreamTableEnvironment.create(senv);

• 定义一个水位线

//泛型指定为Userproduct对象//指定乱序时间两秒//复写方法extractTimestampWatermarkStrategy<Userproduct> watermarkStrategy =WatermarkStrategy.<Userproduct>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(newSerializableTimestampAssigner<Userproduct>(){@OverridepubliclongextractTimestamp(Userproduct userproduct,long l){return userproduct.getDate_time()*1000;//需要得到毫秒值}});

• 从socket获取数据,并且把水位线丢进去

//从socket读取数据，指定水位线
        DataStream<Userproduct> userProductDataStream = senv.socketTextStream("hadoop1",9999).map(event -> {
                    String[] arr = event.split(",");
                    Userproduct userproduct = Userproduct.builder().product_id(Integer.parseInt(arr[2])).buyer_name(arr[3]).date_time(Long.valueOf(arr[0])).price(Double.valueOf(arr[1])).build();return userproduct;
                }).assignTimestampsAndWatermarks(watermarkStrategy);

• 将流式数据给转换成为动态表

Tabletable= tEnv.fromDataStream(userProductDataStream,
                $("product_id"),//跟上字段
                $("buyer_name"),
                $("price"),
                $("date_time").rowtime());//通过调用rowtime来指定event_time为准

• 执行flinkSQL的窗口函数

这边TUMBLE指的是定义滚动窗口，select后面的窗口字段要在groupby也要出现

Table resultTable = tEnv.sqlQuery("select product_id,max(price),TUMBLE_START(date_time,INTERVAL '5' second)  as winstart   "+"from  "+ table +" GROUP BY product_id , TUMBLE(date_time, INTERVAL '5' second)");//间隔5秒

• 完整代码

publicclassFlinkSQLTumbEvtWindowTime{publicstaticvoidmain(String[] args){//构建表执行环境StreamExecutionEnvironment senv =StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);StreamTableEnvironment tEnv =StreamTableEnvironment.create(senv);//定义一个水位线//泛型指定为Userproduct对象//指定乱序时间两秒//复写方法extractTimestampWatermarkStrategy<Userproduct> watermarkStrategy =WatermarkStrategy.<Userproduct>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(newSerializableTimestampAssigner<Userproduct>(){@OverridepubliclongextractTimestamp(Userproduct userproduct,long l){return userproduct.getDate_time()*1000;//需要得到毫秒值}});//从socket读取数据，指定水位线DataStream<Userproduct> userProductDataStream = senv.socketTextStream("hadoop1",9999).map(event ->{String[] arr = event.split(",");Userproduct userproduct =Userproduct.builder().product_id(Integer.parseInt(arr[2])).buyer_name(arr[3]).date_time(Long.valueOf(arr[0])).price(Double.valueOf(arr[1])).build();return userproduct;}).assignTimestampsAndWatermarks(watermarkStrategy);//将流式数据给转换成为动态表Table table = tEnv.fromDataStream(userProductDataStream,
                $("product_id"),//跟上字段
                $("buyer_name"),
                $("price"),
                $("date_time").rowtime());//通过调用rowtime来指定event_time为准//执行flink的sql程序Table resultTable = tEnv.sqlQuery("select product_id,max(price),TUMBLE_START(date_time,INTERVAL '5' second)  as winstart   "+"from  "+ table +" GROUP BY product_id , TUMBLE(date_time, INTERVAL '5' second)");

        resultTable.execute().print();}}

• 最后打印看看

       resultTable.execute().print();

• 看下结果:每隔5秒是一个窗口，每个两秒往前滚动一次

2.3 滑动窗口sql

//使用HOP，滑动大小2秒，窗口大小4秒Table resultTable = tEnv.sqlQuery("select product_id,max(price),HOP_START(date_time,INTERVAL '2' second,INTERVAL '4' second) as winstart  "+"from "+ table +" group by product_id ,HOP(date_time,INTERVAL '2' second,INTERVAL '4' second)");

2.4 会话窗口sql

Table resultTable = tEnv.sqlQuery("select product_id,max(price) , SESSION_START(date_time,INTERVAL '5' second ) as winstart  "+"from "+ table +" group by product_id ,SESSION(date_time,INTERVAL '5' second)");

3 over窗口的使用

3.1 语法

select 分析函数 over (partitionBy 字段 orderby 字段 <开窗范围>) from group by 

• 开窗范围

--范围间隔,例如开窗范围选择当前行之前 1 小时的数据
RANGE BETWEENINTERVAL'1'HOURPRECEDINGANDCURRENTROW--行间隔，例如开窗范围选择当前行之前的 5 行数据(含当前行6行数据)ROWSBETWEEN5PRECEDINGANDCURRENTROW

3.2 案例

1. 数据

1666620609,44,1,白天磊
1666620610,45,1,陈智渊
1666620611,46,1,崔钰轩
1666620612,47,1,吴鹏飞
1666620613,48,1,毛明辉
1666620614,49,1,侯弘文
1666620615,50,1,曾伟祺
1666620616,51,1,郝瑞霖
1666620617,52,1,陆熠彤
1666620618,53,1,余弘文
1666620619,54,1,石哲瀚
1666620620,55,1,任擎苍
1666620621,56,1,卢文轩
1666620622,57,1,吕晋鹏
1666620623,58,1,罗晟睿
1666620624,59,1,周建辉
1666620625,60,1,卢皓轩
1666620626,61,1,沈煜城
1666620627,62,1,万鑫鹏
1666620628,63,1,沈思远

1. 需求与实现

• 使用Over窗口按event-time排序有界向前5s开窗，求取最大值以及平均金额

Table resultTable = tEnv.sqlQuery(
                "select product_id,max(price)" + "OVER w AS max_price, " +"avg(price) OVER w AS avg_price   "+"from "+ table +" WINDOW w AS ( PARTITION BY product_id ORDER BY date_time RANGE BETWEEN INTERVAL '5' second PRECEDING AND CURRENT ROW) ");

当然也可以这么写

Table resultTable = tEnv.sqlQuery(
                "select product_id,max(price)" + "OVER ( PARTITION BY product_id ORDER BY date_time RANGE BETWEEN INTERVAL '5' second PRECEDING AND CURRENT ROW)AS max_price, " +"avg(price) OVER ( PARTITION BY product_id ORDER BY date_time RANGE BETWEEN INTERVAL '5' second PRECEDING AND CURRENT ROW) AS avg_price   "+"from "+ table   

• 使用Over窗口按event-time排序有界向前3条数据，求最大金额以及平均金额

Table resultTable = tEnv.sqlQuery("select product_id,max(price) "+"OVER w AS max_price, avg(price) OVER w AS avg_price  "+" from "+ table 
                +" WINDOW w AS ( PARTITION BY product_id ORDER BY date_time ROWS BETWEEN  3 PRECEDING AND CURRENT ROW) ");