Flink处理watermark未来时间问题

出现原因

通常这种时间超前的数据是由于机器的时间有问题,如机器时间不同/埋点的日志时间错误,然后采集上来的数据使用的那个时间可能就会比当前时间超前了
【举例:处理机器时间戳100s,上报的日志写了2000s】

网上解决方法

1、从kafka读取出来之前先做过滤

在 Flink 从 Kafka 中消费数据后就进行 filter部分这种数据(可以获取到时间后和当前时间相比一下,如果超前或者超前多久就把这条数据丢掉,设置的超前 5分钟以上的数据就丢失),就不让进入后面生成水印,这样就不会导致因为水印过大而导致你后面的问题。

2、自定义水印生成策略

在生成水印的地方做判断,如果采集上来的数据的时间远大于当前时间(比如超过 5分钟)那么就不把当前水印的时间设置为数据的时间,而是用当前系统的时间代替
【自定义水印生成策略,正常为 Math.max(timestamp//当前时间, currentMaxTimestamp)//最大水印】

3.重写Kafka反序列化方法

在kafka反序列化的时候,判断kafka中日期字段的值,如果超过当前时间太多,则丢弃,或者重置为当前时间(重置其实可能导致正常数据丢失)。
【重写Kafka反序列化方法,kafka消息中消息的时间戳和上报的日志中的时间戳进行比较】

4、等待直到未来事件(不可行)

自定义一个watermark,当时间大于当前时间太多的时候,不更新当前的watermark,这样在watermark达到这条未来时间的时间点后,也会将这条数据纳入窗口计算,这种其实是比较理想的。但是这种没有完全的测试,感觉数据会一直存放在内存中,不知道会不会引起其他问题。
【这个相当于将等处理的机器到达2000s再处理前面的,容易造成数据堆积,不可行】

上一篇:FTP客户端、SSH客户端


下一篇:Flink源码解析算子与算子之间的数据是根据什么规则分发到下游的?让你更好的理解Flink流式处理