Fl9nk 水位线简介
作者:互联网
什么是水位线
在事件时间语义下,我们不依赖系统时间,而是基于数据自带的时间戳去定义了一个时钟,用来表示当前时间的进展。于是每个并行子任务都会有一个自己的逻辑时钟,它的前进是靠数据的时间戳来驱动的。但在分布式系统中,这种驱动方式又会有一些问题。因为数据本身在处理转换的过程中会变化,如果遇到窗口聚合这样的操作,其实是要攒一批数据才会输出一个结果,那么下游的数据就会变少,时间进度的控制就不够精细了。另外,数据向下游任务传递时,一般只能传输给一个子任务(除广播外),这样其他的并行子任务的时钟就无法推进了。例如一个时间戳为9整的数据到来,当前任务的时钟就已经是9点了;处理完当前数据要发送到下游,如果下游任务是一个窗口计算,并行度为3,那么接收到这个数据的子任务,时钟也会进展到9点,9点结束的窗口就可以关闭进行计算了;而另外两个并行子任务则时间没有变化,不能进行窗口计算。所以我们应该把时钟也以数据的形式传递出去,告诉下游任务当前时间的进展;而且这个时钟的传递不会因为窗口聚合之类的运算而停滞。一种简单的想法是,在数据流中加入一个时钟标记,记录当前的事件时间;这个标记可以直接广播到下游,当下游任务收到这个标记,就可以更新自己的时钟了。由于类似于水流中用来做标志的记号,在Flink中,这种用来衡量事件时间(Event Time)进展的标记,就被称作“水位线”(Watermark)。具体实现上,水位线可以看作一条特殊的数据记录,它是插入到数据流中的一个标记点,主要内容就是一个时间戳,用来指示当前的事件时间。而它插入流中的位置,就应该是在某个数据到来之后;这样就可以从这个数据中提取时间戳,作为当前水位线的时间戳了。
如上图所示;每个事件产生的数据,都包含了一个时间戳,我们直接用一个整数表示。这里没有指定单位,可以理解为秒或者毫秒(方便起见,下面讲述统一认为是秒)。当产生于2秒的数据到来之后,当前的事件时间就是2秒;在后面插入一个时间戳也为2秒的水位线,随着数据一起向下游流动。而当5秒产生的数据到来之后,同样在后面插入一个水位线,时间戳也为5,当前的时钟就推进到了5秒。这样,如果出现下游有多个并行子任务的情形,我们只要将水位线广播出去,就可以通知到所有下游任务当前的时间进度了。水位线就像它的名字所表达的,是数据流中的一部分,随着数据一起流动,在不同任务之间传输。这看起来非常简单;接下来我们就进一步探讨一些复杂的状况
1. 有序流中的水位线
在理想状态下,数据应该按照它们生成的先后顺序、排好队进入流中;也就是说,它们处理的过程会保持原先的顺序不变,遵守先来后到的原则。这样的话我们从每个数据中提取时间戳,就可以保证总是从小到大增长的,从而插入的水位线也会不断增长、事件时钟不断向前推进。实际应用中,如果当前数据量非常大,可能会有很多数据的时间戳是相同的,这时每来一条数据就提取时间戳、插入水位线就做了大量的无用功。而且即使时间戳不同,同时涌来的数据时间差会非常小(比如几毫秒),往往对处理计算也没什么影响。所以为了提高效率,一般会每隔一段时间生成一个水位线,这个水位线的时间戳,就是当前最新数据的时间戳。所以这时的水位线,其实就是有序流中的一个周期性出现的时间标记。
这里需要注意的是,水位线插入的“周期”,本身也是一个时间概念。在当前事件时间语义下,假如我们设定了每隔100ms生成一次水位线,那就是要等事件时钟推进100ms才能插入;但是事件时钟本身的进展,本身就是靠水位线来表示的——现在要插入一个水位线,可前提又是水位线要向前推进100ms,这就陷入了死循环。所以对于水位线的周期性生成,周期时间是指处理时间(系统时间),而不是事件时间。
2. 乱序流中的水位线
有序流的处理非常简单,看起来水位线也并没有起到太大的作用。但这种情况只存在于理想状态下。我们知道在分布式系统中,数据在节点间传输,会因为网络传输延迟的不确定性,导致顺序发生改变,这就是所谓的“乱序数据”。这里所说的“乱序”(out-of-order),是指数据的先后顺序不一致,主要就是基于数据的产生时间而言的。如图6-7所示,一个7秒时产生的数据,生成时间自然要比9秒的数据早;但是经过数据缓存和传输之后,处理任务可能先收到了9秒的数据,之后7秒的数据才姗姗来迟。这时如果我们希望插入水位线,来指示当前的事件时间进展,又该怎么做呢? 最直观的想法自然是跟之前一样,我们还是靠数据来驱动,每来一个数据就提取它的时间戳、插入一个水位线。不过现在的情况是数据乱序,所以有可能新的时间戳比之前的还小,如果直接将这个时间的水位线再插入,我们的“时钟”就回退了——水位线就代表了时钟,时光不能倒流,所以水位线的时间戳也不能减小。解决思路也很简单:我们插入新的水位线时,要先判断一下时间戳是否比之前的大,否则就不再生成新的水位线,也就是说,只有数据的时间戳比当前时钟大,才能推动时钟前进,这时才插入水位线。如果考虑到大量数据同时到来的处理效率,我们同样可以周期性地生成水位线。这时只需要保存一下之前所有数据中的最大时间戳,需要插入水位线时,就直接以它作为时间戳生成新的水位线
这样做尽管可以定义出一个事件时钟,却也会带来一个非常大的问题:我们无法正确处理“迟到”的数据。在上面的例子中,当9秒产生的数据到来之后,我们就直接将时钟推进到了9秒;如果有一个窗口结束时间就是9秒(比如,要统计0~9秒的所有数据),那么这时窗口就应该关闭、将收集到的所有数据计算输出结果了。但事实上,由于数据是乱序的,还可能有时间戳为7秒、8秒的数据在9秒的数据之后才到来,这就是“迟到数据”(late data)。它们本来也应该属于0~9秒这个窗口,但此时窗口已经关闭,于是这些数据就被遗漏了,这会导致统计结果不正确。如果用之前我们类比班车的例子,现在的状况就是商品不是按照生产时间顺序到来的,所以有可能出现这种情况:9点生产的商品已经到了,我们认为已经到了9点,所以直接发车;但是可能还会有8点59分59秒生产的商品迟到了,没有赶上这班车。那怎么解决这个问题呢?其实我们有很多生活中的经验。假如是一个团队出去团建,那肯定希望每个人都不能落下;如果有人因为堵车没能准时到车上,我们可以稍微等一会儿。9点发车,我们可以等到9点10分,等人都到齐了再出发。当然,实际应用的网络环境不可能跟北京的交通一样堵,所以不需要等那么久,或许只要等一两秒钟就可以了。具体在商品班车的例子里,我们可以多等2秒钟,也就是当生产时间为9点零2秒的商品到达,时钟推进到9点零2秒,这时就认为所有8点到9点生产的商品都到齐了,可以正式发车。不过这样相当于更改了发车时间,属于“违规操作”。为了做到形式上仍然是9点发车,我们可以更改一下时钟推进的逻辑:当一个商品到达时,不要直接用它的生产时间作为当前时间,而是减上两秒,这就相当于把车上的逻辑时钟调慢了。这样一来,当9点生产的商品到达时,我们当前车上的时间是8点59分58秒,还没到发车时间;当9点零2秒生产的商品到达时,车上时间刚好是9点,这时该到的商品都到齐了,准时发车就没问题了。回到上面的例子,为了让窗口能够正确收集到迟到的数据,我们也可以等上2秒;也就是用当前已有数据的最大时间戳减去2秒,就是要插入的水位线的时间戳,如图6-10所示。这样的话,9秒的数据到来之后,事件时钟不会直接推进到9秒,而是进展到了7秒;必须等到11秒的数据到来之后,事件时钟才会进展到9秒,这时迟到数据也都已收集齐,0~9秒的窗口就可以正确计算结果了。
如果仔细观察就会看到,这种“等2秒”的策略其实并不能处理所有的乱序数据。比如22秒的数据到来之后,插入的水位线时间戳为20,也就是当前时钟已经推进到了20秒;对于10~20秒的窗口,这时就该关闭了。但是之后又会有17秒的迟到数据到来,它本来应该属于10~20秒窗口,现在却被遗漏丢弃了。那又该怎么办呢?既然现在等2秒还是等不到17秒产生的迟到数据,那自然我们可以试着多等几秒,也就是把时钟调得更慢一些。最终的目的,就是要让窗口能够把所有迟到数据都收进来,得到正确的计算结果。对应到水位线上,其实就是要保证,当前时间已经进展到了这个时间戳,在这之后不可能再有迟到数据来了。下面是一个示例,我们可以使用周期性的方式生成正确的水位线。
第一个水位线时间戳为7,它表示当前事件时间是7秒,7秒之前的数据都已经到齐,之后再也不会有了;同样,第二个、第三个水位线时间戳分别为12和20,表示11秒、20秒之前的数据都已经到齐,如果有对应的窗口就可以直接关闭了,统计的结果一定是正确的。这里由于水位线是周期性生成的,所以插入的位置不一定是在时间戳最大的数据后面。另外需要注意的是,这里一个窗口所收集的数据,并不是之前所有已经到达的数据。因为数据属于哪个窗口,是由数据本身的时间戳决定的,一个窗口只会收集真正属于它的那些数据。也就是说,上图中尽管水位线W(20)之前有时间戳为22的数据到来,10~20秒的窗口中也不会收集这个数据,进行计算依然可以得到正确的结果。关于窗口的原理,我们会在后面继续展开讲解。
水位线特点
- 水位线是插入到数据流中的一个标记,可以认为是一个特殊的数据
- 水位线主要的内容是一个时间戳,用来表示当前事件时间的进展
- 水位线是基于数据的时间戳生成的
- 水位线的时间戳必须单调递增,以确保任务的事件时间时钟一直向前推进
- 水位线可以通过设置延迟,来保证正确处理乱序数据
- 一个水位线Watermark(t),表示在当前流中事件时间已经达到了时间戳t, 这代表t之前的所有数据都到齐了,之后流中不会出现时间戳t’≤t的数据
标签:窗口,简介,插入,水位,时间,Fl9nk,数据,时钟 来源: https://www.cnblogs.com/wdh01/p/16055971.html