摘要：
- 随着Flink1.5的发布，FlinkSQL 流批一体更加成熟与完善。
- Flink SQL可以替代Flink DataStream实现窗口计算。
1 Flink?翻滚窗口适用场景
1.1?定义
将数据依据固定的窗口度对无界数据流进行切片。
1.2?特点
时间对、窗口长度固定、event无重叠。
1.3?适用场景
BI统计(计算各个时间段的指标)
2 Flink SQL窗口编程模型
Table table = input
?.window([GroupWindow w].as("w")) //?定义一个group window并指定别名w
?.groupBy($("w")) //?按照窗口w对table进行分组
?.select($("b").sum()); // select子句指定返回的列和聚合运算(非键控(key)的window)
Table table = input
?.window([GroupWindow w].as("w")) //?定义一个group window并指定别名w
?.groupBy($("w"), $("a")) //?按照属性a和窗口w对table进行分组（键控的window）
?.select($("a"), $("b").sum()); // select子句指定返回的列和聚合运算
在select子句中，我们还可以返回Window的属性：start，end，rowtime
注意：基于时间的窗口是左闭右开的，例如从9点开始创建一个1个时的窗口，则start为09:00:00.000，end为10:00:00.000，rowtime为09:59:59.999。时间戳正好等于end的event是不会被分组到这个窗口的。
Table table = input
?.window([GroupWindow w].as("w")) //?定义一个group window并指定别名w
?.groupBy($("w"), $("a")) //?按照属性a和窗口w对table进行分组（键控的window）
?.select($("a"), $("w").start(), $("w").end(), $("w").rowtime(), $("b").count());
?//select?句返回字段a、窗口的开始时间戳、窗口的结束时间戳、窗口的时间戳，b字段的count
注意：我们到底取哪个时间戳是由业务决定的，一般是start。
3 Flink SQL滚动窗口实现
3.1?滚动窗口参数
滚动窗口通过Tumble类来定义，三个方法：
3.2?基于EventTime的滚动窗口实现
package com.bigdata.chap05;
import com.bigdata.entity.TempSensorData;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
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.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import java.time.Duration;
import java.time.ZoneId;
import static org.apache.flink.table.api.Expressions.*;
/**
?*?基于事件时间的滚动窗口
?*/
public class FlinkTableTumbleWinBaseEventTime {
????public static void main(String[] args) throws Exception {
????????//1、获取Stream执行环境
????????StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
????????//2、创建表执行环境
????????StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
????????System.out.println(tEnv.getConfig().getLocalTimeZone());
????????env.setParallelism(1);
????????//3、读取数据并提取时间戳指定水印生成策略
????????WatermarkStrategy<TempSensorData> watermarkStrategy = WatermarkStrategy
????????????????.<TempSensorData>forBoundedOutOfOrderness(Duration.ofSeconds(2))
????????????????.withTimestampAssigner(new SerializableTimestampAssigner<TempSensorData>() {
????????????????????@Override
????????????????????public long extractTimestamp(TempSensorData element, long recordTimestamp) {
????????????????????????return element.getTp()*1000;
????????????????????}
????????????????});
????????DataStream<TempSensorData> tempSensorData = env.socketTextStream("hadoop1", 8888)
????????????????.map(event -> {
????????????????????String[] arr = event.split(",");
????????????????????return TempSensorData
????????????????????????????.builder()
????????????????????????????.sensorID(arr[0])
????????????????????????????.tp(Long.parseLong(arr[1]))
????????????????????????????.temp(Integer.parseInt(arr[2]))
????????????????????????????.build();
????????????????}).assignTimestampsAndWatermarks(watermarkStrategy);
????????tEnv.getConfig().setLocalTimeZone(ZoneId.of("Asia/Shanghai"));
????????//4、流转换为动态表
????????Table table = tEnv.fromDataStream(tempSensorData,
????????????????$("sensorID"),
????????????????$("tp"),
????????????????$("temp"),
????????????????$("evTime").rowtime(),//新增evTime字段为rowtime
????????????????$("pt").proctime()
????????);
????????//table.execute().print();
????????//5、自定义窗口并计算
????????Table result = table.window(Tumble
//窗口大小为2s
????????????????.over(lit(2).second())
//按照eventTime排序
????????????????.on($("evTime"))
????????????????.as("w"))
//按照sensorID和窗口分组
????????????????.groupBy($("sensorID"), $("w"))
//统计每个窗口的平均气温
????????????????.select($("sensorID"), $("temp").avg().as("avgTemp"));
????????//6、打印
????????result.execute().print();
????}
}
3.3?测试数据集
在hadoop1节点上面打开nc服务：nc -lk 8888 ?，输入以下数据集测试运行
s-5,1645085900,14
s-5,1645085901,17
s-5,1645085902,22
s-5,1645085903,7
s-5,1645085904,21
s-5,1645085905,23
s-5,1645085906,8
s-5,1645085907,32
s-5,1645085908,15
s-5,1645085909,9
如果能基于EventTime按照时间窗口统计出每个传感器的平均气温，则说明Flink SQL翻滚窗口实现成功。
