最近在跟一个Flink实时数仓的项目,目前还在规划阶段.
今天吃兰州拉面的时候和一个神州实习的同学(后端人,在前两天的文章中出现过--老演员了)交流病情.
他问了我个很严肃的问题 你知道WaterMark吗?
我一下拍在桌子上,跟他讲了半个钟Flink的事件时间、窗口、WaterMark
结果他最后说......他问的是Kafka的WaterMark
目录
基本概念
在事件时间体系中, 时间的进度依赖于数据本身, 和任何设备的时间无关.? 事件时间程序必须制定如何产生Event Time Watermarks(水印) . 在事件时间体系中, 水印是表示时间进度的标志(作用就相当于现实时间的时钟).
没看懂不要怕,咱来慢慢说.
首先提到水印,我们就要先明确一个点:只有使用事件时间,才需要WaterMark
时间语义
在Flink的时间语义中,有三个概念:①处理时间? ? ? ? ? ②注入时间? ? ? ? ? ?③事件时间
在1.12之前默认的时间语义是处理时间, 从1.12开始, Flink内部已经把默认的语义改成了事件时间
①处理时间? ? ? ? 执行操作的设备的时间
②注入时间? ? ? ? 数据注入Flink source的时间
③事件时间? ? ? ? 这个事件发生的时间
①②都比较好理解,就不多说了,至于③? ? ?什么叫这个事件发生的时间?????
事件时间
在event进入Flink之前, 事件时间通常被嵌入到了event中, 一般作为这个event的时间戳存在.
举个栗子:假如说我们接收到的Event都是JavaBean,里面有一个ts属性,用来保存这条信息发生时的时间戳,那么这个时间戳就可以当作我们的事件时间来使用
就好比某个网游出了个升级活动,光棍节前一天升到111级奖励一张好人卡,然后你就高高兴兴跑去做任务了,做完了任务,服务器卡了,你的任务完成消息直到光棍节当天才被处理,如果网游使用的是处理时间,也就是使用的是执行操作的设备的时间,那你就领不到好人卡了!血亏!?
但是网游不能让你抓到把柄喷他,他用了事件时间,就用ts这个字段来表示完成事件发生时间,这样不管你的任务完成记录什么时间被处理,你都能领到那张属于自己的好人卡,wow!
WaterMark的几种类型
再说一遍:在事件时间体系中, 水印是表示时间进度的标志(作用就相当于现实时间的时钟).
随着数据流的流动 WaterMark只会保持不变或者变大,永远不会变小(回到过去)
1.单调递增的WaterMark
?好!就用上面这个栗子来聊一下,红色方框中的是上游的DStream,泛型为<WaterSensor>是个JavaBean,我们调用DStream的assignTimestampsAndWatermarks()方法来声明这个DStream中的WaterMark,该方法需要的参数是WatermarkStrategy(水印策略)类型的,所以我们直接用如图红色横线表明的方法,来声明一个单调递增的WaterMark,至于这个绿色方框的地方,因为这个forMonotonousTimestamps()方法在源码中我们可以发现它是个泛型方法,所以要在前面写明泛型!
最后就是在绿色横线这里我们用匿名内部类实现了一个好像是叫序列化时间戳声明器的一个接口(英文水平有限)
在这个匿名内部类中,实现了一个方法,返回的long类型就是事件时间的值,在上图中,也就是我把ts这个属性的值赋值给了事件时间? ? ? ? ? ? ? ? ? ?
-------也可以说是赋值给了WaterMark,但是据我所知源码中应该是有减1这个概念,因为窗口时间涉及到左闭右开,但是这么去理解,问题也不大
这个是需要实现的方法的源码解释,表明返回的时间戳是独立于任何一个时区和日期的.这是专属于你这个程序的时间逻辑
2.支持乱序的WaterMark
?
这时候就有小朋友要问了:你这WaterMark刚刚不是说只能不变或者增大吗,这怎么还乱序了??
首先我们可以看到除了红色框中的代码,其他地方几乎一摸一样.
红色框中的就是另一种WatermarkStrategy,参数传入的是三秒,也就是允许三秒的乱序
举个栗子,原来的窗口时间为[0,5),也就是WaterMark到了5,这个窗口就会被关闭.但是设置了三秒的乱序时间之后,直到WaterMark达到5+3时,该窗口才会被关闭.
那么这个乱序时间在开发中应该怎么去设置?设置大了容易同时维护过多窗口影响性能,设置小了会达不到预期效果.
所以就还是一句老话,"有经验的人告诉我"
或者就是试运行一段时间,拿出能满足所有数据都被对应窗口接收的乱序时间作为参数.
3.自定义WatermarkStrategy
?自定义WaterMark策略有两种路线,一种是周期型的,一种是间歇型的.
对应着源码可以看出来,如果你想用间歇型生成WaterMark的话,就在onEvent里书写代码,反之也一样
周期型的是默认200ms生成一次,这是写在源码里的? ? ? ? ? ? 间歇型是一条event执行一次
所以有以下结论
周期型适合短时间处理多条event
间歇型适合长时间处理少量event
其他
当程序开始时,WaterMark会被设置为Long的最小值,以保证它不会丢数据
当程序关闭时,WaterMark会被设置为Long的最大值,以保证它大到足以关闭所有已经开启的窗口
多并行度下WaterMark的传递
WaterMark是随着流在程序中流动的,有以下三条结论需要知道
1.多并行度的条件下,向下游传递WaterMark的时候是以广播的方式传递的
2.总是以最小的那个WaterMark为准(木桶原理)
3.当watermark值没有增长的时候不会向下游传递?
(完)
若有不足,欢迎指正
?