Flink水印机制（watermark）

悠悠 2023-08-17 15:16 138阅读 0赞

Flink流处理时间方式

*  EventTime 
    
    时间发生的时间，例如：点击网站上的某个链接的时间
 *  IngestionTime 
    
    某个Flink节点的source operator接收到数据的时间，例如：某个source消费到kafka中的数据 
 *  ProcessingTime 
    
    某个Flink节点执行某个operation的时间，例如：timeWindow接收到数据的时间 
    
    ![421906-20190831181851261-1386371070.png][]

设置Flink流处理的时间类型

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

问题

![421906-20190831181857244-596072746.png][]

1. 使用时间窗口来统计10分钟内的用户流量

2. 有一个时间窗口

*  开始时间为：2017-03-19 10:00:00
 *  结束时间为：2017-03-19 10:10:00

3. 有一个数据，因为网络延迟

*  事件发生的时间为：2017-03-19 10: 10 :00
 *  但进入到窗口的时间为：2017-03-19 10:10: 02 ，延迟了2秒中

4. 时间窗口并没有将 59 这个数据计算进来，导致数据统计不正确

这种处理方式，根据消息进入到window时间，来进行计算。在网络有延迟的时候，会引起计算误差。

水印（watermark）

水印就是一个时间戳，可以给每个消息添加一个 允许一定延迟 的时间戳

*  窗口可以继续计算一定时间范围内延迟的消息 
 *  添加水印后，窗口会等 5 秒，再执行计算。若超过5秒，则舍弃。 
 *  窗口执行计算时间由 水印时间 来触发，当接收到消息的 watermark >= endtime ，触发计算 
    
       
    
    ![421906-20190831181903291-185575618.png][]

Flink提供添加水印的API

<table style="border-collapse:collapse;background:#FFFFFF;"> 
 <tbody> 
 <tr> 
 <td style="border-top:none;border-left:none;border-bottom:none;border-right:none;">&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;val watermarkData: DataStream[Message] = &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;clicklogDataStream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[Message] &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;{ &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;var currentTimestamp: Long = 0L &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;val maxDelayTime = 5000L &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;var watermark: Watermark = null &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;// 获取当前的水印 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;override def getCurrentWatermark = { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;watermark = new Watermark(currentTimestamp - maxDelayTime) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;watermark &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;// 时间戳抽取操作 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;override def extractTimestamp(t: Message, l: Long) = { &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;val timeStamp = t.timestamp &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;currentTimestamp = Math.max(timeStamp, currentTimestamp) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;currentTimestamp &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;} &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;})</td> 
 </tr> 
 </tbody> 
</table>

转载于:https://www.cnblogs.com/starzy/p/11439997.html

[421906-20190831181851261-1386371070.png]: /images/20230810/0cb156aafcf4433db2bcc527bb739b7b.png
[421906-20190831181857244-596072746.png]: /images/20230810/7445a593cf01482db87dc8562c3537c8.png
[421906-20190831181903291-185575618.png]: /images/20230810/9aec97f6ada74c78920a2d0eb1e75a0e.png