Flink的waterMark概念解释

Posted 2022-03-30 月疯

Stream数据中Time（时间）可以分为三种：

Event Time:事件真正产生的时间，即业务系统真正产生日志的时间，它通常由事件中的时间戳描述。
Ingestion time:事件进入Flink的时间；
Processing Time：事件被处理时，当前的系统时间，默认使用的就是这个词。

我们统计Event time窗口某个时间段内产生的日志。存在一个问题：

kafka出来的数据有可能是乱序的，也有可能是延迟的。

这种情况下我们就定义出了watermark得到概念。

为什么需要watermark？
在flink当中，当我们基于event time 进行窗口计算时，由于数据存在乱序和延迟到来的问题，即最先进入窗口计算的数据不一定是在业务上最先产生的数据，所以我们需要提供一种机制，保证对应窗口内的数据已经到达，这样才能触发窗口计算，这个机制就是watermark机制。
watermark是flink为了处理event time窗口计算提出的一种机制，本质上就是一个时间戳，代表着比这个时间早的事件已经全部进入到相应的窗口，后续不会在有比这个时间小的事件出现，基于这个前提我们才有可能将event time窗口视为完整并触发窗口的计算。

我们知道，流处理从事件产生，到流经source，再到operator，中间是有一个过程和时间的。虽然大部分情况下，流到operator的数据都是按照事件产生的时间顺序出现，但是也不排除由于网络延迟等原因，导致乱序的产生，特别是使用kafka的话，多个分区的数据无法保证有序。所以在进行window计算的时候，我们又不能无限期的等下去，必须要有个机制来保证一个特定的时间后，必须触发window去进行计算了。这个特别的机制，就是watermark，watermark是用于处理乱序事件的，watermark可以翻译为水位线！

注意：基于event time窗口计算，存在一个问题就是延迟和乱序，才提出了watermark。

wantermark的三种应用场景：

1、有序流中的wartermark

2、无序流中的watermark

 

3、多并行度中的watermark

  

watermark触发eventTime窗口计算的条件：

只要水印water的时间大于等于窗口的结束时间，并且窗口内有数据存在，就会触发对应窗口的计算；除此之外，如果flink配置allowedLateness参数，只要水印watermark的时间小于等于窗口的结束时间加上allowedLateness参数时间，将会重新触发对应窗口的计算。

水印watermark的生成方式
通常在接收到source的数据后，应该立即生成watermark，然后watermark随着数据流向传输，在flink当中提供了俩种生成watermark的方式：

方式1： periodic Watermark
周期性的（基于一定时间间隔或者达到一定的记录条数）产生一个watermark，默认是100ms，在实际的生产环境当中一般使用这种方式；

class GenerateWaterMark implements AssignerWithPeriodicWatermarks<MyEvent>
    //maxOutOfOrderness表示允许数据的最大乱序时间
    Long maxOutOfOrderness = 3500L;
    Long currentMaxTimestamp =0L;
    //获取watermark的水位线
    @Nullable
    @Override
    public Watermark getCurrentWatermark() 
        return new Watermark(currentMaxTimestamp-maxOutOfOrderness);
    
    //从数据本身提取EventTime
    @Override
    public long extractTimestamp(Event event, long l) 
        long timestamp=event.getEventTime();
        currentMaxTimestamp=Math.max(timestamp,currentMaxTimestamp);
        return timestamp;
    

方式2 ：Punctuated Watermark
数据流中每一个递增的eventTime都会产生一个watermark，这种方式因为会产生大量的watermark所以容易对下游算子造成压力，所以只有在实时性要求非常高的场景才会选择这种方式。