Windowing TVF
在Flink1.13版本之后出现的替代之前的Group window的产物，官网描述其 is more powerful and effective
//TVF 中的tumble滚动窗口
//tumble(table sensor,descriptor(et),interval '5' second ):作为一张表存在
//特别注意！！！！
//如果在sql中使用了tumble窗口，则一定需要group by,而且group by后一定有window_start,window_end两个字段
sql实现TVF的tumble窗口实现
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 Demo1_Window_TableAPI_Tumble {
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 中的tumble滚动窗口
//tumble(table sensor,descriptor(et),interval '5' second ):作为一张表存在
//特别注意！！！！
//如果在sql中使用了tumble窗口，则一定需要group by,而且group by后一定有window_start,window_end两个字段
tabEnv.sqlQuery("select" +
" window_start,window_end,id," +
"sum(vc) sum_vc" +
" from table (tumble(table sensor,descriptor(et),interval '5' second ))" +
" group by window_start,window_end,id ")
.execute()
.print();
}
}
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();
}
}
sql实现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;标签：WaterSensor,窗口,FlinkSQL,window,import,table,TVF,sensor,Windowing From： https://www.cnblogs.com/CYan521/p/16848901.html
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();
}
}