flink WaterMark之TumblingEventWindow
2024-09-08 09:02:16
1、WaterMark,翻译成水印或水位线,水印翻译更抽象,水位线翻译接地气。
watermark是用于处理乱序事件的,通常用watermark机制结合window来实现。 流处理从事件产生,到流经source,再到operator,中间是有一个过程和时间的。虽然大部分情况下,流到operator的数据都是按照事件产生的时间顺序来的,但是也不排除由于网络、背压等原因,导致乱序的产生(out-of-order或者说late element)。 但是对于迟到或者乱序的元素,我们又不能无限期的等下去,必须要有个机制来保证一个特定的时间后,必须触发window进行计算。这个特别的机制,就是watermark。触发时间遵循自然时间以及左闭右开原则。
正常有序流:watermark实际上与event的时间戳重合
乱序流:watermark用于触发窗口计算,也就是水印不到,即使流数据已经落入多个窗口也不会触发,如果水印到了,该窗口的数据即使没到也会触发计算,迟到的数据缺省将被抛弃。
2、TumblingEventWindow 窗口结合WaterMark,用代码验证一下有序和乱序的流。
从socket里接收文本,文本以对子(时间戳 +文本)出现,字段分隔符是空格,行分隔符是“\n”,对收到的文本以10秒滚动窗口给文本计数。
有序情况下:watermark是0,也就是不延时接收数据。
乱序情况下:watermark是3s,延时3秒触发窗口计算。
code:
public class TumblingEventWindowExample {
public static void main(String args[]) throws Exception{
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// env.setParallelism(1);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); DataStream<String> socketStream = env.socketTextStream("192.168.31.10",9000);
DataStream<Tuple2<String,Long>> resultStream = socketStream
//Time.seconds(3)有序的情况修改成0
.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(3)) {
@Override
public long extractTimestamp(String element) {
long eventTime = Long.parseLong(element.split(" ")[0]);
System.out.println(eventTime);
return eventTime;
}
})
.map(new MapFunction<String, Tuple2<String,Long>>() {
@Override
public Tuple2<String,Long> map(String value) throws Exception {
return Tuple2.of(value.split(" ")[1],1L);
}
}).keyBy(0)
.window(TumblingEventTimeWindows.of(Time.seconds(10)))
.reduce(new ReduceFunction<Tuple2<String,Long>>() {
@Override
public Tuple2<String, Long> reduce(Tuple2<String, Long> value1, Tuple2<String, Long> value2) throws Exception {
return new Tuple2<>(value1.f0,value1.f1+value2.f1);
}
});
resultStream.print(); env.execute();
}
}
2.1 有序的情况,watermark为0s
第一个窗口:
10000
11000
12000
13000
14000
19888
13000
20000
1> (b,2)
3> (a,5)
时间戳20000触发第一个窗口计算,实际上19999也会触发,因为左闭右开原则,20000这个时间戳并不会在第一个窗口计算,第一个窗口是[10000-20000),第二个窗口是[20000-30000),以此类推。
第二个窗口:
10000
11000
12000
13000
14000
19888
13000
20000
1> (b,2)
3> (a,5)
11000
12000
21000
22000
29999
3> (a,3)
1> (b,1)
第一个窗口触发计算后,后续来的11000,12000这两条数据被抛弃,29999直接触发窗口计算,并且本身也属于第二个窗口,所以也参与计算了。
2.2 watermark为3s的情况
10000
11000
12000
20000
21000
22000
23000
3> (a,2)
1> (b,1)
从数据中可以验证,第一个窗口在20000的时候没有触发计算,而是在23000的时候触发计算,计算内容是第一个窗口[10000,20000),所以20000,21000,22000,23000属于第二个窗口,没有参与计算。
第二个窗口:
10000
11000
12000
20000
21000
22000
23000
3> (a,2)
1> (b,1)
24000
29000
30000
22000
23000
33000
3> (a,6)
1> (b,2)
第二个窗口[20000,30000),它是在33000触发计算,并且,迟到的数据22000,23000也被计算在内(如果这两个数据在水印33000后到达,则会被抛弃),30000和33000是第三个窗口的数据,没有计算在内。
最新文章
- Java中创建对象的5种方式
- html5快速入门(四)—— JavaScript
- 0909 a newbeginning
- U3D杂记
- Android笔记:ninepatch
- 基于.NET的微软ORM框架视频教程(Entity Framework技术)
- n条直线最多能将一个平面分成多少部分?
- jquery绑定事件失效的情况(转)
- web前端网页特效大全导航列表
- Redis pipeline and list
- POJ 2253 Frogger(floyd)
- 关于IE8中使用Jquery load方法无法正常加载页面
- maven项目中找不到Maven Dependencies解决办法
- mysql 局域网同事之间直接用客户端访问
- php 二级级联菜单
- 汽车之家汽车品牌Logo信息抓取 DotnetSpider实战[三]
- 【学习笔记Part 2● MySQL】
- 【DirectX12】第六章-练习
- acl使用示例
- HDUOJ----(2612)Find a way
热门文章
- [b0036] python 归纳 (二一)_多进程数据共享和同步_服务进程Manager
- PHP代码篇(三)--常用方法
- Wireshark抓包笔录--之指定IP地址筛选捕获结果
- Html学习之十五(盒模型)
- Html学习之九(CSS选择器的使用--位置选择器)
- 使用CMD命令部署.NetCore程序到IIS
- Paper | Beyond a Gaussian Denoiser: Residual Learning of Deep CNN for Image Denoising
- 小白专场-堆中的路径-c语言实现
- C# HTTP系列 HttpWebRequest 与 HttpWebResponse
- iOS:获取一周7天的日期(年-月-日-星期)