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

<>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 ANDCURRENT 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) ");

• « 上一篇：软件安装（Tomcate，Mysql，lrzsz）
• » 下一篇：String详解，创建多少对象问题