Flink 水位线(Watermark)个人总结

一、水位线定义

        在Flink中，用于衡量事件时间希望的标记，被称为水位线。
        可以看作一条特别的数据记录，它是插入到数据流中的一个标记点。主要内容就是一个时间戳，用来指示当前的事件时间
        自身理解：本质上就是一个时间戳，表示比这个时间早的事件已经全部到达。并且在数据乱序的环境下，通过水位线可以判定出迟到的数据
1）有序流中的水位线

        1.1 理想状态(数据量比力小)，数据按天生的先后顺序进入流中，每条数据产生一个水位线

        1.2 实际应用中，数据量如果非常大，可能同一时间又很多条数据大概相差非常小的时间隔断(几ms)，为了进步效率，一样平常会每隔一段时间天生一个水位线

2）乱序流中的水位线

        由于分布式体系中，数据节点传输，可能由于网络延迟，导致事件时间在后面的数据先发送过来，导致顺序改变

        图中 第7s的数据在第9s后收到
解决办法
        ①：乱序+数据量比力小

        判定一下时间戳是否比之前的大，否则就不再天生新的水位线。如图中第7s的数据来了后发现水位线为9s，这时不更新水位线，知道收到9s之后的数据再更新

        ②：乱序+数据量大

        这时只需要保存一下之前全部数据中最大的时间戳，需要插入水位线时，直接用它作为新的水位线

        ③：乱序+迟到数据

        为了能让窗口正确收集到迟到的数据，我们可以等一段时间，比如2s；也就是用当前已有最大数据的时间戳减2s，就是要插入水位线的时间戳

3） 水位线的特性

        ·水位线是插入到数据流中的一个标记，可以认为是一个特别的数据
        ·水位线主要内容是一个时间戳，用来表示当前事件时间的希望
        ·水位线是基于数据的时间戳天生的
        ·水位线的时间戳必须单调递增，以确保任务的事件事件时钟是不停向前推进的
        ·水位线可以通过设置延迟，来确保正确处置惩罚乱序的数据
4）水位线和窗口的工作原理
        窗口可以看作一个个独立的桶，水位线就是一个标记数据流的时间戳。比如watermark标记为10s，在处置惩罚乱序的数据时可以设置一个延迟，比如说2s。这时，第12s的数据会放在区间为10~20的窗口中，对于水位线，会划分到第一个区间中。到达水位线之后，前一个窗口关闭触发计算

二、天生水位线

2.1 总体原则

        完美的水位线表示这个时间之前的数据已经全部到齐、之后不会再出现。但是如果保证绝对正确，就必须等候足够长的时间，这样会有更多的延迟
        如果希望处置惩罚的更快、实时性更强，那么可以将水位线设置的低一些。但是可能会导致窗口遗漏数据，计算结果禁绝确
2.2 水位线天生策略

        flink的DataStreamAPI中，assignTimestampAndWatermarks()，用于为流中数据分配时间戳，并天生水位线指示事件时间。
        通过此方法，传入WatermarkStrategy参数，是一个接口，负责从流中的数据提取时间戳，另一个是基于时间戳设置水位线
        1. 定义watermark策略 -> 指定是否有序(<WaterSensor>forMonotonousTimestamps() / <WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)))
2.3 Flink内置水位线

        1）有序流内置水位线设置
        主要特点就是时间戳单调增长，永久不会出现事件迟到的问题，直接调用WatermarkStrategy.forMonotonousTimestamps()就可以实现
flink从数据中提取事件时间，每条数据来的时候会带着这个时间作为事件时间。当这个时间超出了水位线，前一个窗口关闭，触发计算

        2）乱序流中内置水位线设置
        需要等候，延迟触发，调用WatermarkStrategy. forBoundedOutOfOrderness()方法

2.4 查察乱序数据流水位线源码

乱序：

默认的当前最大事件为Long的最小值+传入的延迟事件
onPeriodicEmit：周期性的发射watermark，默认200ms。
升序：       

内置的水位线都是周期天生的，默认是200ms。
有序流：watermark = 当前最大的事件时间 - 1ms
乱序流：watermark = 当前最大的事件时间 - 延迟时间 - 1ms
2.5 水位线的传递

        多并行度，对于一个子任务，上游多个往它这边发送数据，可能它们的水位线希望不一样。对于一个task收到多个watermark，这时以最小的watermark作为基准。由于最小的大家肯定都处置惩罚过
        ps： 水位线设置空闲等候
        在多个上游并行任务中，如果此中一个没有数据，由于当前task是以最小的那个作为当前任务的事件时钟，就会导致当前task水位线无法推进，就可能导致窗口无法触发。(并行度为2，延迟等候时间为2s，窗口巨细为10s，7s，10s，13s，仍然不会触发窗口，由于此时最小并行度为10 - 2 = 8)这时可以设置空闲等候时间
2.6 迟到数据的处置惩罚

迟到数据：事件时间小于watermark。 可能会有迟到数据无法进入窗口(由于窗口已关闭，触发计算)
2.6.1 推进水印延迟

        推迟体系时间的推进，设置一个延迟时间，保证窗口计算被延迟执行，为后续乱序的数据争取时间进入窗口。通过定义watermark策略中，设置乱序等候时间,有一个Duration
        在watermark中指定乱序等候的时间
        当前数据最大的事件时间 - 延迟时间 大于水位线了，窗口关闭，触发计算

2.6.2 设置窗口延迟关闭

        通过.window().allowedLateness()设置推迟窗口关闭时间。当触发了窗口计算时，会先计算当前结果，但是此时不会关闭窗口。
        以后每来一条数据，就触发一次窗口增量计算，知道watermark超过了窗口结束时间+推迟时间，然后关闭窗口
        eg：当前窗口巨细为10s，窗口推迟关闭时间为3s
                现在来了7s、13s数据，窗口会触发计算，水位线为10s
                迟到数据第5s来了之后会增量进行计算，水位线仍然为10
                直到水位线 > 窗口结束时间+推迟时间(10+3)才会关窗

2.6.3 关窗后的数据使用测输出流吸收

乱序迟到数据的总结：
1）watermark中指定 乱序等候的时间
2）如果开窗，可以设置窗口允许迟到
3）关窗后的迟到数据，放入侧输出流
问题：如果watermark等候3s，窗口允许迟到2s，为什么不直接设置watermark等候5s 大概 窗口允许迟到5s呢？
        1. watermark等候时间一样平常不会设置太高太大 => 会影响计算延迟。窗口关闭触发计算变延后
                如果3s => 窗口第一次触发计算和输出， 13s的数据来 触发计算
                如果5s => 窗口第一次触发计算和输出，15s的数据来 触发计算
        2. 窗口允许迟到 是对 大部分迟到数据的处置惩罚，尽量修正结果，让结果准确
                如果只设置允许迟到5s，那么就会导致 频仍重新输出
        =》 1. watermark设置不算特别大的，一样平常是秒级，在乱序和延迟取舍
                2. 设置肯定的窗口允许迟到，只考虑大部分的迟到数据，
                3. 极端小部分迟到很久的数据就不管了，大概放入侧输出流，获取到之后再处置惩罚
三、一些常见问题

1. 水位线起什么作用

用来衡量时间时间进度的机制，是不断递增的一个时间戳，表示当前时间点，全部小于该时间戳的事件均到达。
2. 水位线和窗口巨细的关系

a：窗口的触发是基于水位线的，当水位线的时间戳大于或等于某个窗口的结束时间，这个窗口就会关闭并触发计算
b：水位线的记录是独立于窗口计算的，它基于事件时间和延迟进行设置。默认是200ms。当解析失事件时间后，会根据当前最大的事件时间 - 延迟时间 - 1ms作为水位线。



免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。