出现原因
通常这种时间超前的数据是由于机器的时间有问题,如机器时间不同/埋点的日志时间错误,然后采集上来的数据使用的那个时间可能就会比当前时间超前了
【举例:处理机器时间戳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再处理前面的,容易造成数据堆积,不可行】