devbox
Monodyee
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

flink水位线

作者：Monodyee | 2024-04-20 02:00:16

赞

踩

flink水位线

目录

一、什么是水位线

1》有序流中的水位线

2》乱序流中的水位线

3》水位线特性

二、水位线和窗口的工作原理

1》窗口

三、 生成水位线

1》生成水位线的总体原则

2》水位线生成策略

3》 Flink内置水位线

四、自定义水位线生成器

1》周期性水位线生成器（Periodic Generator）      

2》断点式水位线生成器（Punctuated Generator）

3》在数据源中发送水位线

五、水位线的传递

六、迟到数据的处理 

1》推迟水印推进

2》设置窗口延迟关闭

3》使用侧流接收迟到的数据

一、什么是水位线

        在Flink中，用来衡量事件时间进展的标记，就被称为“水位线(Watermark)”。

        水位线可以看作一条特殊的数据记录，它是插入到数据流中的一个标记点，主要内容就是一个时间戳，用来指示当前的事件时间。而它插入流中的位置，就应该是在某个数据到来之后；这样就可以从这个数据中提取时间戳，作为当前水位线的时间戳。

1》有序流中的水位线

        （1）理想状态（数据量小），数据应该按照生成的先后顺序进入流中，每条数据生产一个水位线；      

        （2）实际应用中，如果当前数量非常大，且同时涌来的数据时间差会非常小（比如几毫秒），往往对处理计算也没什么影响。所以为了提高效率，一般会每隔一段时间生成一个水位线。

2》乱序流中的水位线

        （1）乱序流中的水位线：在发布式系统中，数据在节点间传输，会因为网络传输的不确定行导致顺序发生改变，这就是所谓的“乱序数据”。

        （2）乱序+数据量小：靠数据来驱动，每来一个数据就提取它的时间戳、插入一个水位线。不过现在的情况是数据乱序，所以插入新的水位线时，要先判断一下时间戳是否比之前的大，否则就不再生成新的水位线。也就是说：只要数据的时间戳比前时钟大，才能推动时钟前进，这时才插入水位线。

        （3）乱序+数据量大：如果考虑到大量数据同时到来的处理效率，同样可以周期性地生成水位线。这时只需要保存一下所有数据种的最大时间戳，需要插入水位线时就直接以它作为时间戳生产新的水位线。

        （4）乱序+迟到数据：我们无法正确处理“迟到”的数据。为了让窗口能够正确收集到迟到的数据，我们也可以等上一段时间，比如2秒；也就是用当前已有数据的最大时间戳减去2秒，就是要插入的水位线的时间戳。这样的话，9秒的数据到来之后，事件时钟不会直接推进到9秒，而是进展到了7秒；必须等到11秒的数据到来之后，事件时钟才会进展到9秒，这时迟到数据也都已收集齐，0~9秒的窗口就可以正确计算结果了。

        现在可以知道水位线就代表了当前的事件时间时钟，而且可以在数据的时间戳基础上加一些延迟来保证不丢数据，这一点对于乱序流的正确处理非常重要

3》水位线特性

        1、水位线是插入到数据流中的一个标记，可以认为是一个特殊的数据；
        2、水位线主要的内容是一个时间戳，用来表示当前事件时间的进展；
        3、水位线是基于数据的时间戳生成的；
        4、水位线的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进；
        5、水位线可以通过设置延迟，来保证正确处理乱序数据；
        6、一个水位线Watermark(T)，表示在当前流中事件时间已经达到了时间戳 t ，这代表t之前的所有数据都到齐了，之后流中不会出现时间戳 t ’ ≤ t 的数据；
        水位线是Flink流处理中保证结果正确性的核心机制，它往往会跟窗口一起配合，完成对乱序数据的正确处理。

二、水位线和窗口的工作原理

1》窗口

        （1）误解：在Flik中，窗口就是用来处理无界流的核心.我们很容易把窗口想象成一个固定位置的“框”，数据源源不断地流过来，到某个时间点窗口该关闭了，就停止收集数据、触发计算并输出结果。

             注意:为了明确数据划分到哪一个窗口，定义窗口都是包含起始时间、不包含结束时间的，用数学符号表示就是一个左闭右开的区间，例如0~10秒的窗口可以表示为[0.10]，这里单位为秒。

        （2）正确理解：在Flink中，窗口其实并不是一个“框”，应该把窗口理解成一个“桶”。在Flink中，窗口可以把流切割成有限大小的多个“存储桶”(桶)：每个数据都会分发到对应的桶中，当到达窗口结束时间时，就对每个桶中收集的数据进行计算处理。

        注意：Flink中窗口并不是静态准备好的，而是动态创建——当有落在这个窗口区间范围的数据达到时，才创建对应的窗口。另外，这里我们认为到达窗口结束时间时，窗口就触发计算并关闭，事实上“触发计算”和“窗口关闭”两个行为也可以分开，这部分内容我们会在后面详述

三、 生成水位线

1》生成水位线的总体原则

        完美的水位线是“绝对正确”的，也就是一个水位线一旦出现，就表示这个时间之前的数据已经全部到齐、之后再也不会出现了。不过如果要保证绝对正确，就必须等足够长的时间，这会带来更高的延迟。

        如果我们希望处理得更快、实时性更强，那么可以将水位线延迟设得低一些。这种情况下，可能很多迟到数据会在水位线之后才到达，就会导致窗口遗漏数据，计算结果不准确。当然，如果我们对准确性完全不考虑、一味地追求处理速度，可以直接使用处理时间语义，这在理论上可以得到最低的延迟。

        所以Flink中的水位线，其实是流处理中对低延迟和结果正确性的一个权衡机制，而且把控制的权力交给了程序员，我们可以在代码中定义水位线的生成策略。

2》水位线生成策略

        在Flink的DataStream API中，有一个单独用于生成水位线的方法：.assignTimestampsAndWatermarks()，它主要用来为流中的数据分配时间戳，并生成水位线来指示事件时间。具体使用如下：

  1. DataStream<Event> stream = env.addSource(new ClickSource());
  2.  
  3. DataStream<Event> withTimestampsAndWatermarks = 
  4. stream.assignTimestampsAndWatermarks(<watermark strategy>);

        说明：WatermarkStrategy作为参数，这就是所谓的“水位线生成策略”。WatermarkStrategy是一个接口，该接口中包含了一个“时间戳分配器”TimestampAssigner和一个“水位线生成器”WatermarkGenerator。

  1. public interface WatermarkStrategy<T> 
  2.     extends TimestampAssignerSupplier<T>,
  3.             WatermarkGeneratorSupplier<T>{
  4.  
  5.     // 负责从流中数据元素的某个字段中提取时间戳，并分配给元素。时间戳的分配是生成水位线的基础。
  6.     @Override
  7.     TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context);
  8.  
  9.     // 主要负责按照既定的方式，基于时间戳生成水位线
  10.     @Override
  11.     WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
  12. }

3》 Flink内置水位线

        （1）有序流中内置水位线设置

        对于有序流，主要特点就是时间戳单调增长，所以永远不会出现迟到数据的问题。这是周期性生成水位线的最简单的场景，直接调用WatermarkStrategy.forMonotonousTimestamps()方法就可以实现。

  1. public class WatermarkMonoDemo {
  2.     public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4.         env.setParallelism(1);
  5.  
  6.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  7.                 .socketTextStream("hadoop102", 7777)
  8.                 .map(new WaterSensorMapFunction());
  9.  
  10.         // TODO 1.定义Watermark策略
  11.         WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
  12.                 // 1.1 指定watermark生成：升序的watermark，没有等待时间
  13.                 .<WaterSensor>forMonotonousTimestamps()
  14.                 // 1.2 指定 时间戳分配器，从数据中提取
  15.                 .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  16.                     @Override
  17.                     public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  18.                         // 返回的时间戳，要 毫秒
  19.                         System.out.println("数据=" + element + ",recordTs=" + recordTimestamp);
  20.                         return element.getTs() * 1000L;
  21.                     }
  22.                 });
  23.  
  24.         // TODO 2. 指定 watermark策略
  25.         SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);
  26.  
  27.  
  28.         sensorDSwithWatermark.keyBy(sensor -> sensor.getId())
  29.                 // TODO 3.使用 事件时间语义 的窗口
  30.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)))
  31.                 .process(
  32.                         new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  33.  
  34.                             @Override
  35.                             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  36.                                 long startTs = context.window().getStart();
  37.                                 long endTs = context.window().getEnd();
  38.                                 String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
  39.                                 String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
  40.  
  41.                                 long count = elements.spliterator().estimateSize();
  42.  
  43.                                 out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
  44.                             }
  45.                         }
  46.                 )
  47.                 .print();
  48.  
  49.         env.execute();
  50.     }
  51. }

        （2）乱序流中内置水位线设置

        由于乱序流中需要等待迟到数据到齐，所以必须设置一个固定量的延迟时间。这时生成水位线的时间戳，就是当前数据流中最大的时间戳减去延迟的结果，相当于把表调慢，当前时钟会滞后于数据的最大时间戳。调用WatermarkStrategy. forBoundedOutOfOrderness()方法就可以实现。这个方法需要传入一个maxOutOfOrderness参数，表示“最大乱序程度”，它表示数据流中乱序数据时间戳的最大差值；如果我们能确定乱序程度，那么设置对应时间长度的延迟，就可以等到所有的乱序数据了。

  1. public class WatermarkOutOfOrdernessDemo {
  2.     public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4.         env.setParallelism(1);
  5.  
  6.  
  7.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  8.                 .socketTextStream("hadoop102", 7777)
  9.                 .map(new WaterSensorMapFunction());
  10.  
  11.  
  12.         // TODO 1.定义Watermark策略
  13.         WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
  14.                 // 1.1 指定watermark生成：乱序的，等待3s
  15.                 .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  16.                 // 1.2 指定 时间戳分配器，从数据中提取
  17.                 .withTimestampAssigner(
  18.                         (element, recordTimestamp) -> {
  19.                             // 返回的时间戳，要 毫秒
  20.                             System.out.println("数据=" + element + ",recordTs=" + recordTimestamp);
  21.                             return element.getTs() * 1000L;
  22.                         });
  23.  
  24.         // TODO 2. 指定 watermark策略
  25.         SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);
  26.  
  27.  
  28.         sensorDSwithWatermark.keyBy(sensor -> sensor.getId())
  29.                 // TODO 3.使用 事件时间语义 的窗口
  30.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)))
  31.                 .process(
  32.                         new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  33.  
  34.                             @Override
  35.                             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  36.                                 long startTs = context.window().getStart();
  37.                                 long endTs = context.window().getEnd();
  38.                                 String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
  39.                                 String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
  40.  
  41.                                 long count = elements.spliterator().estimateSize();
  42.  
  43.                                 out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
  44.                             }
  45.                         }
  46.                 )
  47.                 .print();
  48.  
  49.         env.execute();
  50.     }
  51. }

四、自定义水位线生成器

1》周期性水位线生成器（Periodic Generator）      

        周期性生成器一般是通过onEvent()观察判断输入的事件，而在onPeriodicEmit()里发出水位线。下面是一段自定义周期性生成水位线的代码：

  1. import com.bw.bean.Event;
  2. import org.apache.flink.api.common.eventtime.*;
  3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  4.  
  5. // 自定义水位线的产生
  6. public class CustomPeriodicWatermarkExample {
  7.  
  8.     public static void main(String[] args) throws Exception {
  9.  
  10.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  11.  
  12.         env
  13.                 .addSource(new ClickSource())
  14.                 .assignTimestampsAndWatermarks(new CustomWatermarkStrategy())
  15.                 .print();
  16.  
  17.         env.execute();
  18.     }
  19.  
  20.     public static class CustomWatermarkStrategy implements WatermarkStrategy<Event> {
  21.  
  22.         @Override
  23.         public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
  24.  
  25.             return new SerializableTimestampAssigner<Event>() {
  26.  
  27.                 @Override
  28.                 public long extractTimestamp(Event element，long recordTimestamp) {
  29.                     return element.timestamp; // 告诉程序数据源里的时间戳是哪一个字段
  30.                 }
  31.             };
  32.         }
  33.  
  34.         @Override
  35.         public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
  36.             return new CustomBoundedOutOfOrdernessGenerator();
  37.         }
  38.     }
  39.  
  40.     public static class CustomBoundedOutOfOrdernessGenerator implements WatermarkGenerator<Event> {
  41.  
  42.         private Long delayTime = 5000L; // 延迟时间
  43.         private Long maxTs = -Long.MAX_VALUE + delayTime + 1L; // 观察到的最大时间戳
  44.  
  45.         @Override
  46.         public void onEvent(Event event，long eventTimestamp，WatermarkOutput output) {
  47.             // 每来一条数据就调用一次
  48.             maxTs = Math.max(event.timestamp，maxTs); // 更新最大时间戳
  49.         }
  50.  
  51.         @Override
  52.         public void onPeriodicEmit(WatermarkOutput output) {
  53.             // 发射水位线，默认200ms调用一次
  54.             output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
  55.         }
  56.     }
  57. }

        我们在onPeriodicEmit()里调用output.emitWatermark()，就可以发出水位线了；这个方法由系统框架周期性地调用，默认200ms一次。

        如果想修改默认周期时间，可以通过下面方法修改。例如：修改为400ms

env.getConfig().setAutoWatermarkInterval(400L);

2》断点式水位线生成器（Punctuated Generator）

        断点式生成器会不停地检测onEvent()中的事件，当发现带有水位线信息的事件时，就立即发出水位线。我们把发射水位线的逻辑写在onEvent方法当中即可。

3》在数据源中发送水位线

        我们也可以在自定义的数据源中抽取事件时间，然后发送水位线。这里要注意的是，在自定义数据源中发送了水位线以后，就不能再在程序中使用assignTimestampsAndWatermarks方法来生成水位线了。在自定义数据源中生成水位线和在程序中使用assignTimestampsAndWatermarks方法生成水位线二者只能取其一。示例程序如下：

  1. env.fromSource(
  2. kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource"
  3. )

五、水位线的传递

        在流处理中，上游任务处理完水位线、时钟改变之后，要把当前的水位线再次发出，广播给所有的下游子任务。而当一个任务接收到多个上游并行任务传递来的水位线时，应该以最小的那个作为当前任务的事件时钟。

        水位线在上下游任务之间的传递，非常巧妙地避免了分布式系统中没有统一时钟的问题，每个任务都以“处理完之前所有数据”为标准来确定自己的时钟。

        案例：6.3.4.3 中乱序流的watermark，将并行度设为2，观察现象。

        在多个上游并行任务中，如果有其中一个没有数据，由于当前Task是以最小的那个作为当前任务的事件时钟，就会导致当前Task的水位线无法推进，就可能导致窗口无法触发。这时候可以设置空闲等待。

        用5.3.4.6中的自定义分区器，只输入奇数来模拟部分subtask无数据，代码如下：

  1. public class WatermarkIdlenessDemo {
  2.     public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4.  
  5.         env.setParallelism(2);
  6.  
  7.  
  8.         // 自定义分区器：数据%分区数，只输入奇数，都只会去往map的一个子任务
  9.         SingleOutputStreamOperator<Integer> socketDS = env
  10.                 .socketTextStream("hadoop102", 7777)
  11.                 .partitionCustom(new MyPartitioner(), r -> r)
  12.                 .map(r -> Integer.parseInt(r))
  13.                 .assignTimestampsAndWatermarks(
  14.                         WatermarkStrategy
  15.                                 .<Integer>forMonotonousTimestamps()
  16.                                 .withTimestampAssigner((r, ts) -> r * 1000L)
  17.                                 .withIdleness(Duration.ofSeconds(5))  //空闲等待5s
  18.                 );
  19.  
  20.  
  21.         // 分成两组： 奇数一组，偶数一组 ， 开10s的事件时间滚动窗口
  22.         socketDS
  23.                 .keyBy(r -> r % 2)
  24.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)))
  25.                 .process(new ProcessWindowFunction<Integer, String, Integer, TimeWindow>() {
  26.                     @Override
  27.                     public void process(Integer integer, Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
  28.                         long startTs = context.window().getStart();
  29.                         long endTs = context.window().getEnd();
  30.                         String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
  31.                         String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
  32.  
  33.                         long count = elements.spliterator().estimateSize();
  34.  
  35.                         out.collect("key=" + integer + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
  36.  
  37.                     }
  38.                 })
  39.                 .print();
  40.  
  41.  
  42.         env.execute();
  43.     }
  44. }

六、迟到数据的处理 

1》推迟水印推进

        在水印产生时，设置一个乱序容忍度，推迟系统时间的推进，保证窗口计算被延迟执行，为乱序的数据争取更多的时间进入窗口

WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10));

2》设置窗口延迟关闭

         Flink的窗口，也允许迟到数据。当触发了窗口计算后，会先计算当前的结果，但是此时并不会关闭窗口。

        以后每来一条迟到数据，就触发一次这条数据所在窗口计算(增量计算)。直到wartermark 超过了窗口结束时间+推迟时间，此时窗口会真正关闭。

  1. .window(TumblingEventTimeWindows.of(Time.seconds(5)))
  2. .allowedLateness(Time.seconds(3))

        注意：允许迟到只能运用在event time上

3》使用侧流接收迟到的数据

  1. .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
  2. .allowedLateness(Time.seconds(3))
  3. .sideOutputLateData(lateWS)

完整案例代码如下：

  1. public class WatermarkLateDemo {
  2.     public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4.         env.setParallelism(1);
  5.  
  6.  
  7.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  8.                 .socketTextStream("hadoop102", 7777)
  9.                 .map(new WaterSensorMapFunction());
  10.  
  11.         WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
  12.                 .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
  13.                 .withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000L);
  14.  
  15.         SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);
  16.  
  17.  
  18.         OutputTag<WaterSensor> lateTag = new OutputTag<>("late-data", Types.POJO(WaterSensor.class));
  19.  
  20.         SingleOutputStreamOperator<String> process = sensorDSwithWatermark.keyBy(sensor -> sensor.getId())
  21.                 .window(TumblingEventTimeWindows.of(Time.seconds(10)))
  22.                 .allowedLateness(Time.seconds(2)) // 推迟2s关窗
  23.                 .sideOutputLateData(lateTag) // 关窗后的迟到数据，放入侧输出流
  24.                 .process(
  25.                         new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
  26.  
  27.                             @Override
  28.                             public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
  29.                                 long startTs = context.window().getStart();
  30.                                 long endTs = context.window().getEnd();
  31.                                 String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
  32.                                 String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
  33.  
  34.                                 long count = elements.spliterator().estimateSize();
  35.  
  36.                                 out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
  37.                             }
  38.                         }
  39.                 );
  40.  
  41.  
  42.         process.print();
  43.         // 从主流获取侧输出流，打印
  44.         process.getSideOutput(lateTag).printToErr("关窗后的迟到数据");
  45.  
  46.         env.execute();
  47.     }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Monodyee/article/detail/454796
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号