FlinkSQL之Windowing TVF( 二 ) _生活百科

sql实现TVF的滑动窗口
//TVF 中的hop滚动窗口 //hop(table sensor,descriptor(et),interval '2' second,interval '5' second ):作为一张表存在 //first interval :滑动步长， second interval :窗口长度 //特别注意！！！！ // 1.TVF 中滑动窗口的滑动步长与窗口长度必须是整数倍的关系，不然会报错 // 例如：滑动步长为2 ，窗口长度就不能为5 ，可以为6 // 2.如果在sql中使用了hop窗口，则一定需要group by,而且group by后一定有window_start,window_end两个字段 package net.cyan.FlinkSql.TVF; ? import net.cyan.POJO.WaterSensor; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; ? import java.time.Duration; ? import static org.apache.flink.table.api.Expressions.$; ? public class Demo2_Window_TVF_Hop { public static void main(String[] args) { //创建执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //创建表的运行环境 StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env); env.setParallelism(1); DataStream<WaterSensor> waterSensorStream = env.fromElements( new WaterSensor("sensor_1", 1000L, 10), new WaterSensor("sensor_1", 2000L, 20), new WaterSensor("sensor_2", 3000L, 30), new WaterSensor("sensor_1", 4000L, 40), new WaterSensor("sensor_1", 5000L, 50), new WaterSensor("sensor_2", 6000L, 60)) .assignTimestampsAndWatermarks( WatermarkStrategy .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) .withTimestampAssigner((ws, ts) -> ws.getTs()) ? ); //创建table Table table = tabEnv.fromDataStream(waterSensorStream,$("id"),$("ts"),$("vc"),$("et").rowtime()); //创建表 tabEnv.createTemporaryView("sensor",table); //执行sql //TVF 中的hop滚动窗口 //hop(table sensor,descriptor(et),interval '2' second,interval '5' second ):作为一张表存在 //first interval :滑动步长， second interval :窗口长度 //特别注意！！！！ // 1.TVF 中滑动窗口的滑动步长与窗口长度必须是整数倍的关系，不然会报错 // 例如：滑动步长为2 ，窗口长度就不能为5 ，可以为6 // 2.如果在sql中使用了hop窗口，则一定需要group by,而且group by后一定有window_start,window_end两个字段 tabEnv.sqlQuery("select" + "window_start,window_end,id," + "sum(vc) sum_vc" + " from table (hop(table sensor,descriptor(et),interval '2' second,interval '6' second ))" + " group by window_start,window_end,id ") .execute() .print(); ? ? ? } }

FlinkSQL之Windowing TVF( 二 )

经验总结扩展阅读

章太炎简介

煮海螺多长时间

冬天怎么健康减肥冬天健康减肥的方法

钟元凯相亲是哪一期?

|妈妈，我想您了！写给驰援上海抗疫一线妈妈的一封信

德运全脂奶粉冲泡方法,德运奶粉怎么吃？德运奶粉的怎么冲？

中秋节吃柚子的寓意是什么?

南极北极冰川融化会造成什么后果

白酒是伴随着中国历史发展而发展的|酒局上有3种掉价儿的行为，希望你一个也没有

床前明月光打一字

糯米饭怎么做好吃

2023年4月订婚最吉利的日子是哪天

杨超不化妆的杨超越，跟路人有什么区别？大眼睛都消失了

向阳园艺桂花已经到花期了，你家的桂花还没开？可能是没有嫁接！

计算器e是什么意思计算器e指的是什么

油皮换季皮肤干燥，上妆卡纹分享5款好用的保湿面霜，养出水嫩肌肤！

壬辰年是哪一年壬辰年什么命

蓝宝清水|蓝宝清水 | 现代感开放式的混凝土别墅

冷藏室温度多少合适?

OPPO Reno9 Pro和OPPO Reno8 Pro+购机建议 reno8 pro和pro+的区别