Flink窗口迟到机制

6.4 窗口允许迟到的数据

（1）迟到数据概念

迟到数据：时间戳小于 watermark的数据

（2）窗口允许迟到

1） 当 wm >= 窗口最大时间戳时， 会 触发，但是不会关窗
2） 当 窗口最大时间戳 + 允许迟到时间 >= wm >= 窗口最大时间戳, 每来一条迟到数据，都会进入窗口，触发
3） 当 wm >= 窗口最大时间戳 + 允许迟到时间 ， 会关窗， 迟到数据不再进入窗口，不会再触发

（3）代码

无序流watersensor、乱序程度=3、滚动事件窗口大小10s，传入sensor_1,1,1，此时会创建一个[0,10)的窗口，那么触发时间为watermark >= 窗口最大时间 - 1ms，那么watermark = 最大事件时间 - 乱序程度 - 1ms

所以事件时间=13s才会触发执行。

允许迟到时间=2s，所以[0,10)窗口会在[13s-15s)触发，15s会关闭窗口。

public class Flink18_Watermark_allowedLateness {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);env.socketTextStream("hadoop102", 9999).map(new MapFunction<String, WaterSensor>() {@Overridepublic WaterSensor map(String value) throws Exception {String[] dats = value.split(",");return new WaterSensor(dats[0],Long.valueOf(dats[1]),Integer.valueOf(dats[2]));}}).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {@Overridepublic long extractTimestamp(WaterSensor element, long recordTimestamp) {return element.getTs() * 1000L;}})).keyBy(sensor -> sensor.getId()).window(TumblingEventTimeWindows.of(Time.seconds(10)))//TODO 允许迟到时间2s.allowedLateness(Time.seconds(2)).process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {@Overridepublic void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {out.collect("key=" + s + "\n" +"数据为:" + elements + "\n" +"数量条数:" + elements.spliterator().estimateSize() + "\n" +"窗口为:[" + context.window().getStart() + "," + context.window().getEnd() + ")\n" +"=======================================================================\n\n");}}).print();env.execute();}
}

（4）乱序程度(迟到时间)、允许迟到时间

①为什么不直接把乱序程度调大一点？

输出滞后的时间不一样： watermark >= 10s时触发 （不考虑1ms）1） 乱序3s，允许迟到2s  =>  et=13s的时候，触发输出一次结果（滞后3s），后面有迟到的数据，来一条，触发更新一次2） 乱序5s => et = 15s的时候，触发输出一次结果（滞后5s）

Flink窗口迟到机制相关推荐

大数据——Flink Window（窗口）机制
Flink窗口机制 Window(窗口) Tumbling Window(翻滚窗口) Sliding Window(滑动窗口) Sliding Window(滑动窗口)设置Watermark时间 Wi ...
Flink中迟到数据的处理
目录设置水位线延迟时间允许窗口处理迟到数据将迟到数据放入窗口侧输出流总结: 我们知道,所谓的"迟到数据"(late data),是指某个水位线之后到来的数据 ...
Flink的容错机制
文章目录检查点检查点的保存从检查点恢复状态检查点算法检查点配置 1.启动检查点 2.检查点存储 3.其他高级配置保存点 1.保存点的用途 2.使用保存点状态一致性一致性的概念和级别端 ...
【学习笔记 — Flink 处理迟到数据（★）】
Flink 处理迟到数据(★) 处理迟到数据之前首先了解Lambda架构 Lambda架构的实现是:一个批处理器.一个流处理器.流处理器首先实时输出近似正确的结果(因为乱序流,可能导致流处理结果不准确 ...
Flink窗口全解析：三种时间窗口、窗口处理函数使用及案例
我们经常需要在一个时间窗口维度上对数据进行聚合,窗口是流处理应用中经常需要解决的问题.Flink的窗口算子为我们提供了方便易用的API,我们可以将数据流切分成一个个窗口,对窗口内的数据进行处理.本文将 ...
【Flink】迟到数据的处理
窗口在流上的工作方式与批处理不同,因为流通常是无限的,所以不可能计算流中的所有元素,流上的聚合事件则由窗口限定,例如"过去 5 分钟的计数"或"最后 100 个元素的总 ...
Flink对迟到数据的处理的三种方式
** Flink对迟到数据的处理 ** 水位线可以用来平衡计算的完整性和延迟两方面.除非我们选择一种非常保守的水位线策略(最大延时设置的非常大,以至于包含了所有的元素,但结果是非常大的延迟),否则我们 ...
Flink中容错机制完整使用 (第十章)
Flink中容错机制完整使用一.容错机制 1.检查点(Checkpoint) 1. 检查点的保存 1. 周期性的触发保存 2. 保存的时间点 3. 保存的具体流程 2.从检查点恢复状态 (1)重启 ...
【Flink】 Flink JobManager HA 机制的扩展与实现
1.概述转载:Flink 源码阅读笔记(21)- Flink JobManager HA 机制的扩展与实现在 Flink 1.12 中,Flink on Kubernetes 的 Native 部 ...

Flink窗口迟到机制

6.4 窗口允许迟到的数据

（1）迟到数据概念

（2）窗口允许迟到

（3）代码

（4）乱序程度(迟到时间)、允许迟到时间

Flink窗口迟到机制相关推荐

最新文章

热门文章