7.2、如何理解Flink中的水位线(Watermark)

0、版本说明

1、什么是水位线？

2、水位线使用场景？

3、设计水位线主要为了解决什么问题？

4、怎样在flink中生成水位线？

4.1、自定义标记 Watermark 生成器

4.2、自定义周期性 Watermark 生成器

4.3、内置Watermark生成器 - 有序流水位线生成器

4.4、内置Watermark生成器 - 乱序流水位线生成器

4.5、在读取数据源时添加水位线

5、水位线和窗口的关系？

6、水位线在各个算子间的传递

6.1、测试用例 - 不设置 withIdleness 超时时间

6.2、测试用例 - 设置 withIdleness 超时时间

0、版本说明

开发语言：java1.8

Flink版本：1.17

官网链接：官网链接

1、什么是水位线？

Flink中水位线是一条特殊的数据(long timestamp)

它会以时间戳的形式作为一条标识数据插入到数据流中

2、水位线使用场景？

使用事件时间(EventTime)做流式计算任务时，需要根据事件时间生成水位线(Watermark)

通过水位线来触发窗口计算，水位线作为衡量事件时间(EventTime)进展的标识

3、设计水位线主要为了解决什么问题？

设计水位线主要是为了解决实时流中数据乱序和迟到的问题

思考：什么原因造成了数据流的乱序呢？

如今数据采集、数据传输大多都在分布式系统中完成

各个机器节点因为网络和自身性能的原因导致了数据的乱序和迟到

4、怎样在flink中生成水位线？

Flink中支持在数据源和普通DataStream上添加水位线生成策略(WatermarkStrategy)

4.1、自定义标记 Watermark 生成器

标记 Watermark 生成器特点：

每条数据到来后，都会为其生成一条 Watermark

适用场景：

数据量小且数据有序

代码示例：

Step1：自定义标记水位线生成器实现类

// 自定义 标记水位线生成器 实现类
public class PeriodWatermarkGenerator<T> implements WatermarkGenerator<T> {// 每进入一条数据，都会调用一次 onEvent 方法@Override/** 参数说明：*   @event ： 进入到该方法的事件数据*   @eventTimestamp ： 时间戳提取器提取的时间戳* */public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {//发射水位线output.emitWatermark(new Watermark(eventTimestamp));}// 不需要实现@Overridepublic void onPeriodicEmit(WatermarkOutput output) {}
}

Step2：自定义标记性水位线生成策略实现类

// TODO 自定义 标记性水位线生成策略
public class PeriodWatermarkStrategy implements WatermarkStrategy<Tuple2<String, Long>> {// TODO 实例化一个 事件时间提取器@Overridepublic TimestampAssigner<Tuple2<String, Long>> createTimestampAssigner(TimestampAssignerSupplier.Context context) {TimestampAssigner<Tuple2<String, Long>> timestampAssigner = new TimestampAssigner<Tuple2<String, Long>>() {@Overridepublic long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {return element.f1;}};return timestampAssigner;}// TODO 实例化一个 watermark 生成器@Overridepublic WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {return new PeriodWatermarkGenerator<>();}
}

Step3：使用标记性水位线生成策略

// TODO 使用 自定义标记 Watermark 生成器
public class UserPeriodWatermarkStrategy {public static void main(String[] args) throws Exception {// 1.获取执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// 2.将socket作为数据源（开启socket端口: nc -lk 9999）SingleOutputStreamOperator<Tuple2<String, Long>> sourceDataStream = env.socketTextStream("localhost", 9999).map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2 map(String value) throws Exception {return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));}});// 3.为 DataStream 添加水位线生成策略 (使用 自定义WatermarkStrategy 实现类)SingleOutputStreamOperator<Tuple2<String, Long>> assignTimestampsAndWatermarksDs = sourceDataStream.assignTimestampsAndWatermarks(new PeriodWatermarkStrategy());// 4.通过 processFunction实例 查看生成的水位线SingleOutputStreamOperator<String> process = assignTimestampsAndWatermarksDs.process(new ShowProcessFunction());process.print();// 5.触发程序执行env.execute();}
}

查看运行结果：

4.2、自定义周期性 Watermark 生成器

标记 Watermark 生成器特点：

基于处理时间，周期性生成 Watermark

适用场景：

数据量大且可能存在一定程度数据延迟(乱序)

代码示例：

Step1：自定义周期性水位线生成器实现类

// 自定义 周期性水位线生成器
public class PunctuatedWatermarkGenerator<T> implements WatermarkGenerator<T> {// 设置变量，用来保存 当前最大的事件时间private long currentMaxTimestamp;// 设置变量，指定最大的乱序时间(等待时间)private final long maxOutOfOrderness = 0000; // 3 秒@Overridepublic void onEvent(T event, long eventTimestamp, WatermarkOutput output) {// 只更新当前最大时间戳，不再发生水位线if (currentMaxTimestamp < eventTimestamp) currentMaxTimestamp = eventTimestamp;}// 周期性 生成水位线// 每个 setAutoWatermarkInterval 时间，调用一次该方法@Overridepublic void onPeriodicEmit(WatermarkOutput output) {// 发出的 watermark = 当前最大时间戳 - 最大乱序时间output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness));}
}

Setp2：自定义周期性水位线生成策略实现类

// 自定义 周期性水位线生成策略
public class PunctuatedWatermarkStrategy implements WatermarkStrategy<Tuple2<String, Long>> {// TODO 实例化一个 事件时间提取器@Overridepublic TimestampAssigner<Tuple2<String, Long>> createTimestampAssigner(TimestampAssignerSupplier.Context context) {TimestampAssigner<Tuple2<String, Long>> timestampAssigner = new TimestampAssigner<Tuple2<String, Long>>() {@Overridepublic long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {return element.f1;}};return timestampAssigner;}// TODO 实例化一个 watermark 生成器@Overridepublic WatermarkGenerator<Tuple2<String, Long>> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {return new PunctuatedWatermarkGenerator<>();}}

Step3：周期性水位线生成策略

// TODO 使用 自定义周期性 Watermark 生成器
public class UserPunctuatedWatermarkStrategy {public static void main(String[] args) throws Exception {// 1.获取执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// TODO 设置周期性生成水位线的时间间隔(默认为200毫秒)env.getConfig().setAutoWatermarkInterval(3 * 1000L);// 2.将socket作为数据源（开启socket端口: nc -lk 9999）SingleOutputStreamOperator<Tuple2<String, Long>> ds = env.socketTextStream("localhost", 9999).map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2 map(String value) throws Exception {return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));}});// TODO 获取 WatermarkStrategy实例 (方式1：通过 WatermarkStrategy实现类获取)PunctuatedWatermarkStrategy punctuatedWatermarkStrategy = new PunctuatedWatermarkStrategy();// TODO 获取 WatermarkStrategy实例 (方式2：通过 WatermarkStrategy工具类获取) 推荐WatermarkStrategy<Tuple2<String, Long>> punctuatedWatermarkStrategyByUtil = WatermarkStrategy.<Tuple2<String, Long>>forGenerator(context -> new PunctuatedWatermarkGenerator<>()).withTimestampAssigner((event, timestamp) -> event.f1);// 3.使用 自定义水位线策略实例 来提取时间戳&生成水位线SingleOutputStreamOperator<Tuple2<String, Long>> assignTimestampsAndWatermarksDs = ds.assignTimestampsAndWatermarks(punctuatedWatermarkStrategy);// 4.通过 processFunction实例 查看生成的水位线SingleOutputStreamOperator<String> process = assignTimestampsAndWatermarksDs.process(new ShowProcessFunction());process.print();// 3.触发程序执行env.execute();}
}

查看运行结果：

4.3、内置Watermark生成器 - 有序流水位线生成器

有序流水位线生成器特点：

基于处理时间，周期性生成 Watermark，最大乱序时间为0

适用场景：

大数量有序流

代码示例：

// TODO 内置Watermark生成器 - 有序流水位线生成器
public class UserForMonotonousTimestamps {public static void main(String[] args) throws Exception {// 1.获取执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// TODO 设置周期性生成水位线的时间间隔(默认为200毫秒)env.getConfig().setAutoWatermarkInterval(3 * 1000L);// 2.将socket作为数据源（开启socket端口: nc -lk 9999）SingleOutputStreamOperator<Tuple2<String, Long>> sourceDataStream = env.socketTextStream("localhost", 9999).map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2 map(String value) throws Exception {return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));}});// TODO 创建 内置水位线生成策略WatermarkStrategy<Tuple2<String, Long>> watermarkStrategy = WatermarkStrategy.<Tuple2<String, Long>>forMonotonousTimestamps().withTimestampAssigner((element,recordTimestamp) -> element.f1);// 3.使用 内置水位线生成策略SingleOutputStreamOperator<Tuple2<String, Long>> assignTimestampsAndWatermarksDs = sourceDataStream.assignTimestampsAndWatermarks(watermarkStrategy);// 4.通过 processFunction实例 查看生成的水位线SingleOutputStreamOperator<String> process = assignTimestampsAndWatermarksDs.process(new ShowProcessFunction());process.print();// 3.触发程序执行env.execute();}
}

查看运行结果：

4.4、内置Watermark生成器 - 乱序流水位线生成器

乱序流水位线生成器特点：

基于处理时间，周期性生成 Watermark，可以这是最大乱序时间

适用场景：

大数量乱序流

代码示例：

// TODO 内置Watermark生成器 - 乱序流水位线生成器
public class UserForBoundedOutOfOrderness {public static void main(String[] args) throws Exception {// 1.获取执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// TODO 设置周期性生成水位线的时间间隔(默认为200毫秒)env.getConfig().setAutoWatermarkInterval(3 * 1000L);// 2.将socket作为数据源（开启socket端口: nc -lk 9999）SingleOutputStreamOperator<Tuple2<String, Long>> ds = env.socketTextStream("localhost", 9999).map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2 map(String value) throws Exception {return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));}});// TODO 获取 WatermarkStrategy实例WatermarkStrategy<Tuple2<String, Long>> watermarkStrategy = WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(1)) // 设置最大乱序时间为1s.withTimestampAssigner((element,recordTimestamp) -> element.f1);// 3.使用 内置水位线生成策略SingleOutputStreamOperator<Tuple2<String, Long>> assignTimestampsAndWatermarksDs = ds.assignTimestampsAndWatermarks(watermarkStrategy);// 4.通过 processFunction实例 查看生成的水位线SingleOutputStreamOperator<String> process = assignTimestampsAndWatermarksDs.process(new ShowProcessFunction());process.print();// 3.触发程序执行env.execute();}
}

查看运行结果：

4.5、在读取数据源时添加水位线

// 1.获取执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 2.创建 Source 对象
Source source = DataGeneratorSource、KafkaSource...// 3.读取 source时添加水位线
env.fromSource(source, WatermarkStrategy实例, "source name")   .print()
;// 4.触发程序执行
env.execute();

5、水位线和窗口的关系？

窗口什么时候创建？

当窗口内的第一条数据到达时

窗口什么时候触发计算？

当阈值水位线到达窗口时

6、水位线在各个算子间的传递

下游算子 watermark 的计算方式是取所有不同的上游并行数据源 watermark 的最小值

测试代码：

// TODO 测试水位线的传递
public class TransmitWaterMark {public static void main(String[] args) throws Exception {// 1.获取执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(3); // 2.将socket作为数据源（开启socket端口: nc -lk 9999）DataStreamSource<String> source = env.socketTextStream("localhost", 9999);source.partitionCustom(new Partitioner<String>() {@Overridepublic int partition(String key, int numPartitions) {if (key.equals("a")) {return 0;} else if (key.equals("b")) {return 1;} else {return 2;}}}, value -> value.split(",")[0]).map(new MapFunction<String, Tuple2<String, Long>>() {@Overridepublic Tuple2 map(String value) throws Exception {return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));}}).assignTimestampsAndWatermarks(WatermarkStrategy//.<Tuple2<String, Long>>forMonotonousTimestamps().<Tuple2<String, Long>>forGenerator(new PeriodWatermarkStrategy()).withTimestampAssigner((element,recordTimestamp) -> element.f1).withIdleness(Duration.ofSeconds(5))  //空闲等待5s).process(new ShowProcessFunction()).setParallelism(1).print();env.execute();}
}