javaFlink（三十五）Flink的watermark设置以及部分实现类

watermark&＃xff08;水位线&＃xff09;是Flink里边的相当重要的存在&＃xff0c;是Flink处理乱序数据的重要组成。

Flink理论上定义了三种watermark广播机制&＃xff08;【2种生成水印的策略】【如果生成的watermark是null&＃xff0c;或者小于之前的watermark&＃xff0c;则该watermark不会发往下游】&＃xff09;。

首先看看它的组成&＃xff1a;

AssignerWithPunctuatedWatermarks&＃xff08;为每条消息都会尝试生成水印&＃xff09;

AssignerWithPeriodicWatermarks&＃xff08;周期性的生成水印&＃xff0c;不会针对每条消息都生成&＃xff09;&＃xff08;常用&＃xff09;

还有第三种策略是 无为策略&＃xff1a;不设定watermark策略。

这两种策略的源码分析参考https://www.cnblogs.com/ljygz/p/11435243.html

AssignerWithPeriodicWatermarks 实例&＃xff1a;

.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks>() {//定义watermark乱序等待时间2s,即允许数据的最大乱序时间private long maxOutofOrderness &＃61; 2 * 1000;// 观察到的最大时间戳private long currentMaxTs &＃61; Long.MIN_VALUE;&＃64;Nullablepublic Watermark getCurrentWatermark() {//生成具有2s容忍度的水位线return new Watermark(currentMaxTs-maxOutofOrderness );}// 先调用该函数 previousElementTimestamp代表public long extractTimestamp(Tuple2 element, long previousElementTimestamp) {long currentTime &＃61; previousElementTimestamp;// 更新最大的时间戳currentTime &＃61; Math.max(currentMaxTs, currentTime);// 返回记录的时间戳return currentTime;}})

再看看实现接口的几个抽象类&＃xff1a;

关于AscendingTimestampExtractor&＃xff0c;一般是在数据集的时间戳是单调递增的且没有乱序时使用&＃xff0c;该方法使用当前的时间戳生成水位线

.assignTimestampsAndWatermarks(new AscendingTimestampExtractor() {&＃64;Overridepublic long extractAscendingTimestamp(UserBehavior element) {return element.timestamp*1000;}});

 关于BoundedOutOfOrdernessTimestampExtractor&＃xff0c;是在数据集中存在乱序数据的情况下使用&＃xff0c;即数据有延迟(任意新到来的元素与已经到来的时间戳最大的元素之间的时间差)&＃xff0c;这种方式可以接收一个表示最大预期延迟参数&＃xff0c;具体如下&＃xff1a;

.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor(Time.seconds(10)) {&＃64;Overridepublic long extractTimestamp(UserBehavior element) {return element.timestamp*1000;}} );