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

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

FlinkSQL之Windowing TVF( 三 )

经验总结扩展阅读

“离婚|“离婚，我不伺候你一家6口”3个侄子的到来，直接毁了8年的婚姻

三资平台是指什么意思?

怎么治孩子拖拉的毛病

|有些亲戚，真的不值得来往

12岁小学生身高1.9米被高校选中，体育局：征求过本人和家人意见，她1.86米的姐姐也被选中

芒果女演员刘露演的电视有哪些?

dnf进仓库的时候咋没有弹出设置仓库密码

生物威胁的来源有哪些?

棋瑞的寓意是什么？

暗处理的目的,暗处理填有机物可以不

养殖场违建可以强拆吗

实现有效沟通需要做到哪些?

娃娃鱼是指,鲵是什么意思

毛豆搭配啥菜炒好吃,毛豆搭配啥菜炒好吃

丝绸|赵薇又在散发魅力，43岁“挑战”丝绸衣，光腿配鱼嘴鞋更有女人味

大闸蟹蒸熟后怎么保存

结婚|感情从恋爱到婚姻的标准到底应该是什么？

z5次什么车型？

河南是不是沿海城市

美白保湿秋季该怎么做