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 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:窗口长度 //特别注意?。。。?nbsp; //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】

FlinkSQL之Windowing TVF( 三 )

推荐阅读

带梓的女孩名字及寓意

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

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

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

粉色的粘土小玫瑰diy

倚的组词倚的组词组

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

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

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

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

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

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

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

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

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

梦见大鱼吃小鱼

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

OPPO Reno6简单评测 oppo reno6有nfc吗

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

双子座成功人士知多少？