FlinkSql的窗口使用以及运用案例-CFANZ编程社区

1 flinkSQL窗口概述

1.1 窗口定义:

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

1.2 窗口分类：

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

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

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

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

自定义窗口

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 滚动窗口案例

数据

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,沈思远

需求

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

代码实现

定义Userproduct类定义字段

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

构造执行环境

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


        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

定义一个水位线

        //泛型指定为Userproduct对象
        //指定乱序时间两秒
        //复写方法extractTimestamp
        WatermarkStrategy<Userproduct> watermarkStrategy = WatermarkStrategy.<Userproduct>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<Userproduct>() {
                    @Override
                    public long extractTimestamp(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);

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

        Table table = 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秒

完整代码

public class FlinkSQLTumbEvtWindowTime {

    public static void main(String[] args) {

        //构建表执行环境
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(senv);

        //定义一个水位线

        //泛型指定为Userproduct对象
        //指定乱序时间两秒
        //复写方法extractTimestamp
        WatermarkStrategy<Userproduct> watermarkStrategy = WatermarkStrategy.<Userproduct>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<Userproduct>() {
                    @Override
                    public long extractTimestamp(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 BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW	
--行间隔，例如开窗范围选择当前行之前的 5 行数据(含当前行6行数据)
ROWS BETWEEN 5 PRECEDING AND CURRENT ROW

3.2 案例

数据

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,沈思远

需求与实现

使用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) );