案例
通过编写代码，理解水位线的生成，窗口的开始时间、结束时间，触发计算的时机，窗口中的数据
实现
根据 flink1.12.x watermark 源码分析 ，重新实现 BoundedOutOfOrdernessWatermarks 类，打印 watermark
说明
为了便于测试，并行度必须设置为
1，否则打印信息太多，看不出效果MapFunction类，打印接收数据的时间，打印数据，方便查看WindowFunction实现类，打印窗口开始时间、结束时间，和窗口中的数据WatermarkGenerator类，打印 watermark ，方便查看 watermark
flink java代码
package watermarker;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;
import java.util.Properties;
public class SlidingWindow完美验证 {
public static void main(String[] args) throws Exception {
long maxOutOfOrderness = 5;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 定义kafka相关配置项
Properties pro = new Properties();
pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"consumer-group");
pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");
// 内置连接器，创建FlinkKafka消费者，需要指定主题、反序列化、kafka配置
FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer("kafka2flink", new SimpleStringSchema(), pro);
DataStreamSource<String> dss = env.addSource(fkc);
// map解析json数据
SingleOutputStreamOperator<Tuple3<String, Double, Long>> mapSOSO = dss.map(new MyMap());
// 设置watermark最大迟到时间
WatermarkStrategy<Tuple3<String, Double, Long>> watermarkStrategy = new WatermarkStrategy<Tuple3<String, Double, Long>>() {
@Override
public WatermarkGenerator<Tuple3<String, Double, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
return new BoundedOutOfOrdernessWatermarks(Duration.ofSeconds(maxOutOfOrderness));
}
};
//指定watermark事件时间字段
WatermarkStrategy<Tuple3<String, Double, Long>> timestamp = watermarkStrategy // 最大迟到时间
.withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, Double, Long>>() {
@Override
public long extractTimestamp(Tuple3<String, Double, Long> element, long recordTimestamp) {
return element.f2;
}
});
// 设置 watermark
SingleOutputStreamOperator<Tuple3<String, Double, Long>> ataw = mapSOSO
.assignTimestampsAndWatermarks(timestamp);
// keyBy分组
KeyedStream<Tuple3<String, Double, Long>, String> keyByKS = ataw
.keyBy(new KeySelector<Tuple3<String, Double, Long>, String>() {
@Override
public String getKey(Tuple3<String, Double, Long> value) throws Exception {
return value.f0;
}
});
// 设置滚动窗口
WindowedStream<Tuple3<String, Double, Long>, String, TimeWindow> window = keyByKS
.window(TumblingEventTimeWindows.of(Time.seconds(10)));
// 设置窗口函数
SingleOutputStreamOperator<String> apply = window.apply(new MyWindowFunction());
apply.print("result:");
env.execute();
}
// 重新实现 watermark 类，打印系统时间、watermark、事件时间
public static class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {
private long maxTimestamp;
private final long outOfOrdernessMillis;
private long eventTime = 0;
public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();
this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
}
// 取出最大事件时间
@Override
public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
this.eventTime = eventTimestamp;
}
/*
* 最大事件时间 - 最大延迟时间，作为 watermark
* 在这里打印 watermark 时间
*/
@Override
public void onPeriodicEmit(WatermarkOutput output) {
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
long watermark = maxTimestamp - outOfOrdernessMillis - 1;
Date watermarkDate = new Date(watermark);
String watermarkStr = sdf.format(watermarkDate);
Date eventTime = new Date(this.eventTime);
String eventTimeStr = sdf.format(eventTime);
Date date = new Date();
String systime = sdf.format(date);
System.out.println("system time:"+systime+", watermark:"+watermarkStr+", eventTime:"+eventTimeStr);
output.emitWatermark(new Watermark(watermark));
}
}
// 获取 window 开始时间、结束时间，窗口内的数据
public static class MyWindowFunction implements WindowFunction <Tuple3<String, Double, Long>,String,String, TimeWindow>{
@Override
public void apply(String key, TimeWindow window, Iterable<Tuple3<String, Double, Long>> input, Collector<String> out) throws Exception {
StringBuilder stringBuilder = new StringBuilder();
SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
Date windowStartDate = new Date(window.getStart());
Date windowEndDate = new Date(window.getEnd());
String windowStartStr = sdf.format(windowStartDate);
String windowEndStr = sdf.format(windowEndDate);
stringBuilder.append("winow time:["+windowStartStr+","+windowEndStr+")");
stringBuilder.append(", window data:");
for(Tuple3<String, Double, Long> item : input){
stringBuilder.append(item.f1+" , ");
}
out.collect(stringBuilder.toString());
}
}
// 打印接收数据、接收时间
public static class MyMap implements MapFunction<String, Tuple3<String,Double,Long>>{
@Override
public Tuple3<String,Double,Long> map(String data) throws Exception {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date sysDate = new Date();
String sysTime = sdf.format(sysDate);
ObjectMapper om = new ObjectMapper();
JsonNode jsonNode = om.readTree(data);
JsonNode idJN = jsonNode.get("id");
JsonNode tempJN = jsonNode.get("temp");
JsonNode timeJN = jsonNode.get("time");
String id = null;
Double temp = null;
long timeStamp = 0;
if(idJN!=null){
id = idJN.asText();
}
if(tempJN!=null){
temp = tempJN.asDouble();
}
String timeStr = null;
if(timeJN!=null){
timeStr = timeJN.asText();
Date date = sdf.parse(timeStr);
timeStamp = date.getTime();
}
System.out.println("receive:"+Tuple3.of(id,temp,timeStr)+",receive time:"+sysTime);
return Tuple3.of(id,temp,timeStamp);
}
}
}
kafka 生产者
其他配置见 flink1.12.x WaterMark 案例：kafka生产者
主类
package top.malaoshi.kafka;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.Random;
public class Producer循环发送随机延迟 {
final static String TOPIC = "kafka2flink";
public static void main(String[] args) throws InterruptedException {
// 1. 创建 kafka 生产者的配置对象
Properties properties = new Properties();
// 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop1:9092,hadoop2:9092,hadoop3:9092");
// key,value 序列化（必须）：key.serializer，value.serializer
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer");
// StringSerializer.class.getName() 等于 org.apache.kafka.common.serialization.StringSerializer
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
"org.apache.kafka.common.serialization.StringSerializer") ;
// 3. 创建 kafka 生产者对象
KafkaProducer<String, String> kafkaProducer = new KafkaProducer(properties);
Random random = new Random();
double temp = 1;
for(int i=0;i<50;i++) {
// 随机生成7以下的整数，表示 延迟的时间
int delay = random.nextInt(7);
String event1 = EventBuilder.buildJson("1", temp+i, delay);
// 4. 调用 send 方法,发送消息，全部延迟2秒
kafkaProducer.send(new ProducerRecord(TOPIC , event1));
System.out.println("send:" + event1);
Thread.sleep(500);
}
// 5. 关闭资源
kafkaProducer.close();
System.out.println("end...");
}
}
测试
启动zk
登录 hadoop1，执行下面命令：
/program/bin/zk.sh start
启动kafka
登录 hadoop1，执行下面命令：
/program/bin/kafka.sh start
启动 flink
在 idea 中运行 flink 程序
启动生产者、指定主题
执行上面的 kafka 生产者
执行结果
receive:(1,1.0,2022-09-27 11:34:42),system time:2022-09-27 11:34:42
system time:11:34:42, watermark:11:34:36, eventTime:11:34:42
system time:11:34:42, watermark:11:34:36, eventTime:11:34:42
receive:(1,2.0,2022-09-27 11:34:38),system time:2022-09-27 11:34:43
system time:11:34:43, watermark:11:34:36, eventTime:11:34:38
system time:11:34:43, watermark:11:34:36, eventTime:11:34:38
system time:11:34:43, watermark:11:34:36, eventTime:11:34:38
receive:(1,3.0,2022-09-27 11:34:40),system time:2022-09-27 11:34:43
system time:11:34:43, watermark:11:34:36, eventTime:11:34:40
system time:11:34:43, watermark:11:34:36, eventTime:11:34:40
receive:(1,4.0,2022-09-27 11:34:39),system time:2022-09-27 11:34:44
system time:11:34:44, watermark:11:34:36, eventTime:11:34:39
system time:11:34:44, watermark:11:34:36, eventTime:11:34:39
system time:11:34:44, watermark:11:34:36, eventTime:11:34:39
receive:(1,5.0,2022-09-27 11:34:38),system time:2022-09-27 11:34:44
system time:11:34:44, watermark:11:34:36, eventTime:11:34:38
system time:11:34:44, watermark:11:34:36, eventTime:11:34:38
receive:(1,6.0,2022-09-27 11:34:39),system time:2022-09-27 11:34:45
system time:11:34:45, watermark:11:34:36, eventTime:11:34:39
system time:11:34:45, watermark:11:34:36, eventTime:11:34:39
system time:11:34:45, watermark:11:34:36, eventTime:11:34:39
receive:(1,7.0,2022-09-27 11:34:43),system time:2022-09-27 11:34:45
system time:11:34:45, watermark:11:34:37, eventTime:11:34:43
system time:11:34:45, watermark:11:34:37, eventTime:11:34:43
receive:(1,8.0,2022-09-27 11:34:44),system time:2022-09-27 11:34:46
system time:11:34:46, watermark:11:34:38, eventTime:11:34:44
system time:11:34:46, watermark:11:34:38, eventTime:11:34:44
receive:(1,9.0,2022-09-27 11:34:44),system time:2022-09-27 11:34:46
system time:11:34:46, watermark:11:34:38, eventTime:11:34:44
system time:11:34:46, watermark:11:34:38, eventTime:11:34:44
system time:11:34:46, watermark:11:34:38, eventTime:11:34:44
receive:(1,10.0,2022-09-27 11:34:44),system time:2022-09-27 11:34:47
system time:11:34:47, watermark:11:34:38, eventTime:11:34:44
system time:11:34:47, watermark:11:34:38, eventTime:11:34:44
receive:(1,11.0,2022-09-27 11:34:44),system time:2022-09-27 11:34:47
system time:11:34:47, watermark:11:34:38, eventTime:11:34:44
system time:11:34:47, watermark:11:34:38, eventTime:11:34:44
system time:11:34:47, watermark:11:34:38, eventTime:11:34:44
receive:(1,12.0,2022-09-27 11:34:46),system time:2022-09-27 11:34:48
system time:11:34:48, watermark:11:34:40, eventTime:11:34:46
result:> winow time:[11:34:30,11:34:40), window data:2.0 , 4.0 , 5.0 , 6.0 ,
从下面数据可知：
system time:11:34:48, watermark:11:34:40, eventTime:11:34:46
result:> winow time:(11:34:30,11:34:40), window data:2.0 , 4.0 , 5.0 , 6.0 ,
- 窗口时间：
[11:34:30,11:34:40) - 当 watermark 是
11:34:40，大于等于 窗口结束时间11:34:40时，触发计算 - 窗口内的数据：
2.0 , 4.0 , 5.0 , 6.0 ,
窗口时间 解释：
窗口时间：[11:34:30,11:34:40)，数据中的 事件时间 是 [11:34:30,11:34:40) 范围的，都在这个窗口内
注意：窗口时间 左闭右开（左侧包含，右侧不包含），就是说 数据中的事件时间是 11:34:40 的，不在这个窗口内
窗口数据 解释：
由于上面原因，该窗口 包含下面数据（从 接收时间看，已经 超出窗口时间，但仍然会接收并计算）：
receive:(1,2.0,2022-09-27 11:34:38),receive time:2022-09-27 11:34:43
receive:(1,4.0,2022-09-27 11:34:39),receive time:2022-09-27 11:34:44
receive:(1,5.0,2022-09-27 11:34:38),receive time:2022-09-27 11:34:44
receive:(1,6.0,2022-09-27 11:34:39),receive time:2022-09-27 11:34:45
该窗口 不包含以下数据，因为 事件时间 大于等于 11:34:40：
receive:(1,1.0,2022-09-27 11:34:42),system time:2022-09-27 11:34:42
receive:(1,3.0,2022-09-27 11:34:40),system time:2022-09-27 11:34:43
receive:(1,7.0,2022-09-27 11:34:43),system time:2022-09-27 11:34:45
receive:(1,8.0,2022-09-27 11:34:44),system time:2022-09-27 11:34:46
receive:(1,9.0,2022-09-27 11:34:44),system time:2022-09-27 11:34:46
receive:(1,10.0,2022-09-27 11:34:44),system time:2022-09-27 11:34:47
receive:(1,11.0,2022-09-27 11:34:44),system time:2022-09-27 11:34:47
receive:(1,12.0,2022-09-27 11:34:46),system time:2022-09-27 11:34:48