• 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：自定义 标记水位线生成器 实现类

    1. // 自定义 标记水位线生成器 实现类
    2. public class PeriodWatermarkGenerator implements WatermarkGenerator {
    3.  
    4.     // 每进入一条数据，都会调用一次 onEvent 方法
    5.     @Override
    6.     /*
    7.      * 参数说明：
    8.      *   @event ： 进入到该方法的事件数据
    9.      *   @eventTimestamp ： 时间戳提取器提取的时间戳
    10.      * */
    11.     public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
    12.         //发射水位线
    13.         output.emitWatermark(new Watermark(eventTimestamp));
    14.     }
    15.  
    16.     // 不需要实现
    17.     @Override
    18.     public void onPeriodicEmit(WatermarkOutput output) {
    19.     }
    20. }

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

    1. // TODO 自定义 标记性水位线生成策略
    2. public class PeriodWatermarkStrategy implements WatermarkStrategy> {
    3.     // TODO 实例化一个 事件时间提取器
    4.     @Override
    5.     public TimestampAssigner> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
    6.         TimestampAssigner> timestampAssigner = new TimestampAssigner>() {
    7.  
    8.             @Override
    9.             public long extractTimestamp(Tuple2 element, long recordTimestamp) {
    10.                 return element.f1;
    11.             }
    12.         };
    13.         return timestampAssigner;
    14.     }
    15.  
    16.     // TODO 实例化一个 watermark 生成器
    17.     @Override
    18.     public WatermarkGenerator> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
    19.         return new PeriodWatermarkGenerator<>();
    20.     }
    21. }

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

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

    查看运行结果：

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

    标记 Watermark 生成器特点：

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

    适用场景：

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

    代码示例：        

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

    1. // 自定义 周期性水位线生成器
    2. public class PunctuatedWatermarkGenerator implements WatermarkGenerator {
    3.     // 设置变量，用来保存 当前最大的事件时间
    4.     private long currentMaxTimestamp;
    5.     // 设置变量，指定最大的乱序时间(等待时间)
    6.     private final long maxOutOfOrderness = 0000; // 3 秒
    7.  
    8.     @Override
    9.     public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
    10.         // 只更新当前最大时间戳，不再发生水位线
    11.         if (currentMaxTimestamp < eventTimestamp) currentMaxTimestamp = eventTimestamp;
    12.     }
    13.  
    14.     // 周期性 生成水位线
    15.     // 每个 setAutoWatermarkInterval 时间，调用一次该方法
    16.     @Override
    17.     public void onPeriodicEmit(WatermarkOutput output) {
    18.         // 发出的 watermark = 当前最大时间戳 - 最大乱序时间
    19.         output.emitWatermark(new Watermark(currentMaxTimestamp - maxOutOfOrderness));
    20.     }
    21. }

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

    1. // 自定义 周期性水位线生成策略
    2. public class PunctuatedWatermarkStrategy implements WatermarkStrategy> {
    3.     // TODO 实例化一个 事件时间提取器
    4.     @Override
    5.     public TimestampAssigner> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
    6.         TimestampAssigner> timestampAssigner = new TimestampAssigner>() {
    7.  
    8.             @Override
    9.             public long extractTimestamp(Tuple2 element, long recordTimestamp) {
    10.                 return element.f1;
    11.             }
    12.         };
    13.  
    14.         return timestampAssigner;
    15.     }
    16.  
    17.     // TODO 实例化一个 watermark 生成器
    18.     @Override
    19.     public WatermarkGenerator> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
    20.         return new PunctuatedWatermarkGenerator<>();
    21.     }
    22.  
    23. }

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

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

    查看运行结果：

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

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

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

    适用场景：

            大数量有序流

    代码示例：

    1. // TODO 内置Watermark生成器 - 有序流水位线生成器
    2. public class UserForMonotonousTimestamps {
    3.     public static void main(String[] args) throws Exception {
    4.         // 1.获取执行环境
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(1);
    7.  
    8.         // TODO 设置周期性生成水位线的时间间隔(默认为200毫秒)
    9.         env.getConfig().setAutoWatermarkInterval(3 * 1000L);
    10.  
    11.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    12.         SingleOutputStreamOperator> sourceDataStream = env.socketTextStream("localhost", 9999)
    13.                 .map(new MapFunction>() {
    14.                          @Override
    15.                          public Tuple2 map(String value) throws Exception {
    16.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    17.                          }
    18.                      }
    19.                 );
    20.  
    21.         // TODO 创建 内置水位线生成策略
    22.         WatermarkStrategy> watermarkStrategy = WatermarkStrategy.>forMonotonousTimestamps()
    23.                 .withTimestampAssigner((element,recordTimestamp) -> element.f1);
    24.  
    25.         // 3.使用 内置水位线生成策略
    26.         SingleOutputStreamOperator> assignTimestampsAndWatermarksDs = sourceDataStream.assignTimestampsAndWatermarks(watermarkStrategy);
    27.  
    28.         // 4.通过 processFunction实例 查看生成的水位线
    29.         SingleOutputStreamOperator process = assignTimestampsAndWatermarksDs.process(new ShowProcessFunction());
    30.         process.print();
    31.  
    32.         // 3.触发程序执行
    33.         env.execute();
    34.     }
    35. }

    查看运行结果：

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

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

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

    适用场景：

            大数量乱序流

    代码示例：

    1. // TODO 内置Watermark生成器 - 乱序流水位线生成器
    2. public class UserForBoundedOutOfOrderness {
    3.     public static void main(String[] args) throws Exception {
    4.         // 1.获取执行环境
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(1);
    7.  
    8.         // TODO 设置周期性生成水位线的时间间隔(默认为200毫秒)
    9.         env.getConfig().setAutoWatermarkInterval(3 * 1000L);
    10.  
    11.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    12.         SingleOutputStreamOperator> ds = env.socketTextStream("localhost", 9999)
    13.                 .map(new MapFunction>() {
    14.                          @Override
    15.                          public Tuple2 map(String value) throws Exception {
    16.                              return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    17.                          }
    18.                      }
    19.                 );
    20.  
    21.         // TODO 获取 WatermarkStrategy实例
    22.         WatermarkStrategy> watermarkStrategy = WatermarkStrategy
    23.                 .>forBoundedOutOfOrderness(Duration.ofSeconds(1)) // 设置最大乱序时间为1s
    24.                 .withTimestampAssigner((element,recordTimestamp) -> element.f1);
    25.  
    26.         // 3.使用 内置水位线生成策略
    27.         SingleOutputStreamOperator> assignTimestampsAndWatermarksDs = ds.assignTimestampsAndWatermarks(watermarkStrategy);
    28.  
    29.         // 4.通过 processFunction实例 查看生成的水位线
    30.         SingleOutputStreamOperator process = assignTimestampsAndWatermarksDs.process(new ShowProcessFunction());
    31.         process.print();
    32.  
    33.         // 3.触发程序执行
    34.         env.execute();
    35.     }
    36. }

    查看运行结果：

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

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

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

    窗口什么时候创建？

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

    窗口什么时候触发计算？

            当阈值水位线到达窗口时

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

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

    测试代码：

    1. // TODO 测试水位线的传递
    2. public class TransmitWaterMark {
    3.     public static void main(String[] args) throws Exception {
    4.         // 1.获取执行环境
    5.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    6.         env.setParallelism(3); 
    7.  
    8.         // 2.将socket作为数据源（开启socket端口: nc -lk 9999）
    9.         DataStreamSource source = env.socketTextStream("localhost", 9999);
    10.  
    11.         source
    12.                 .partitionCustom(
    13.                         new Partitioner() {
    14.                             @Override
    15.                             public int partition(String key, int numPartitions) {
    16.                                 if (key.equals("a")) {
    17.                                     return 0;
    18.                                 } else if (key.equals("b")) {
    19.                                     return 1;
    20.                                 } else {
    21.                                     return 2;
    22.                                 }
    23.                             }
    24.                         }, value -> value.split(",")[0]
    25.                 )
    26.                 .map(new MapFunction>() {
    27.                     @Override
    28.                     public Tuple2 map(String value) throws Exception {
    29.                         return new Tuple2(value.split(",")[0], Long.parseLong(value.split(",")[1]));
    30.                     }
    31.                 })
    32.                 .assignTimestampsAndWatermarks(
    33.                         WatermarkStrategy
    34.                                 //.>forMonotonousTimestamps()
    35.                                 .>forGenerator(new PeriodWatermarkStrategy())
    36.                                 .withTimestampAssigner((element,recordTimestamp) -> element.f1)
    37.                                 .withIdleness(Duration.ofSeconds(5))  //空闲等待5s
    38.                 )
    39.                 .process(new ShowProcessFunction()).setParallelism(1)
    40.                 .print();
    41.         
    42.         env.execute();
    43.     }
    44. }

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

    现象：如果上游某一个子任务一直没有数据更新，下游算子的水位线一直不会变化

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

    现象：如果上游某一个子任务`在指定时间内`数据更新，下游算子的水位线将不受该子任务最小值的影响

  • 相关阅读:
    AWVS漏洞扫描使用基础与介绍
    打卡记录编程成长/CSDN编程竞赛(第6期)
    Oracle数据库安装及配置
    工控安全PLC固件逆向三
    Nginx搭载负载均衡及前端项目部署
    node.js-模块化
    【HMS】地图标记聚合HWMarkerCluster支持设置聚合计算时网格的像素大小、最大的聚合级别
    最全HTTP/HTTPS面试题整理（三）
    VSCode-常用快捷键
    day04-前台首页、导出项目依赖
  • 原文地址：https://blog.csdn.net/weixin_42845827/article/details/133080491