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

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 :窗口长度 //特别注意?。。。?nbsp; // 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(); ? ? ? } }sql实现TVF的累计窗口
累计窗口的应用：
需求：每天每隔一个小时统计一次当天的pv（浏览量）
流的方式如何解决：
1、用滚动窗口，窗口长度设为1h
2、每天的第一个窗口清除状态，后面的不清，进行状态的累加
或者
用滚动窗口，长度设置为2day
自定义触发器，每隔1小时对窗内的元素计算一次，不关闭窗口
sql的方式如何解决？
直接使用累计窗口cumulate
//TVF 中的cumulate累计窗口 //cumulate(table tableName,descriptor(timecol),step,size):作为一张表存在 //tableName：表名 //timecol:时间属性字段 //step:累计步长，跟滑动步长类似 //size:窗口长度 //特别注意?。。。?nbsp;//1.累计窗口的步长与窗口长度同样是需要整数倍关系 // 2.如果在sql中使用了cumulate窗口，则一定需要group by,而且group by后一定有window_start,window_end两个字段

FlinkSQL之Windowing TVF( 二 )

推荐阅读

带梓的女孩名字及寓意

学生会活动策划方案范文经典学生会活动策划范文

12星座女谁是小野猫谁是小奶猫

月光石手链有什么寓意月光石有什么寓意

粉色的粘土小玫瑰diy

倚的组词倚的组词组

2023武汉中考志愿填报规则详解 2021年武汉中考志愿填报时间

关于深圳市万科教育发展基金会简述深圳市万科教育发展基金会

没有足够的可用内存来运行此程序没有足够的可用内存来运行程序

缓刑是什么意思还要坐牢吗缓刑是什么意思

有一个“撩妹高手”的儿子，是种什么样的体验？

南阳多少人「南阳多少人口2020总人数」

什么是事故车,如何鉴定车子撞到什么程度算事故车

秋季喝什么汤最好秋季喝什么汤最养生

绝地求生刺激战场军团名字怎么改？军团名字修改攻略[多图]

梦见大鱼吃小鱼

文房四宝指什么文房四宝指什么数字

OPPO Reno6简单评测 oppo reno6有nfc吗

怎么判断二手车的真实行驶里程? 要看哪些细节

双子座成功人士知多少？