Apache Flink 漫谈系列 - Watermark

现实题目(乱序)

在先容Watermark相干内容之前我们先抛出一个详细的题目，在现实的流式计较中数据到来的次序对计较功效的正确性有至关重要的影响，好比：某数据源中的某些数据因为某种缘故起因(如：收集缘故起因，外部存储自身缘故起因)会有5秒的延时，也就是在现实时刻的第1秒发生的数据有也许在第5秒中发生的数据之后到来(好比到Window处理赏罚节点)。选详细某个delay的元向来说，假设在一个5秒的Tumble窗口(详见Window先容章节)，有一个EventTime是 11秒的数据，在第16秒时辰到来了。图示第11秒的数据，在16秒到来了，如下图：

那么对付一个Count聚合的Tumble(5s)的window，上面的环境如那里理赏罚才气window2=4，window3=2 呢?

Apache Flink的时刻范例

开篇我们描写的题目是一个很常见的TimeWindow中数据乱序的题目，乱序是相对付变乱发生时刻和达到Apache Flink 现实处理赏罚算子的次序而言的，关于时刻在Apache Flink中有如下三种时刻范例，如下图：

(1)ProcessingTime

ProcessingTime是数据流入到详细某个算子时辰响应的体系时刻。ProcessingTime 有最好的机能和最低的耽误。但在漫衍式计较情形中ProcessingTime具有不确定性，沟通数据流多次运行有也许发生差异的计较功效。

(2)IngestionTime

IngestionTime是数据进入Apache Flink框架的时刻，是在Source Operator中配置的。与ProcessingTime对比可以提供更可猜测的功效，由于IngestionTime的时刻戳较量不变(在源处只记录一次)，统一数据在流经差异窗口操纵时将行使沟通的时刻戳，而对付ProcessingTime统一数据在流经差异窗口算子会有差异的处理赏罚时刻戳。

(3)EventTime

EventTime是变乱在装备上发生时辰携带的。在进入Apache Flink框架之前EventTime凡是要嵌入到记录中，而且EventTime也可以从记录中提取出来。在现实的网上购物订单等营业场景中，大多会行使EventTime来举办数据计较。

开篇描写的题目和本篇要先容的Watermark所涉及的时刻范例均是指EventTime范例。

什么是Watermark

Watermark是Apache Flink为了处理赏罚EventTime 窗口计较提出的一种机制,本质上也是一种时刻戳，由Apache Flink Source可能自界说的Watermark天生器凭证需求Punctuated可能Periodic两种方法天生的一种体系Event，与平凡数据流Event一样流转到对应的下流算子，吸取到Watermark Event的算子以此不绝调解本身打点的EventTime clock。 Apache Flink 框架担保Watermark单调递增，算子吸取到一个Watermark时辰，框架知道不会再有任何小于该Watermark的时刻戳的数据元素到来了，以是Watermark可以看做是汇报Apache Flink框架数据流已经处理赏罚到什么位置(时刻维度)的方法。 Watermark的发生和Apache Flink内部处理赏罚逻辑如下图所示：

Watermark的发生方法

今朝Apache Flink 有两种出产Watermark的方法，如下：

• Punctuated - 数据流中每一个递增的EventTime城市发生一个Watermark。在现实的出产中Punctuated方法在TPS很高的场景下会发生大量的Watermark在必然水平上对下流算子造成压力，以是只有在及时性要求很是高的场景才会选择Punctuated的方法举办Watermark的天生。
• Periodic - 周期性的(一按时距离断可能到达必然的记录条数)发生一个Watermark。在现实的出产中Periodic的方法必需结适时刻和蕴蓄条数两个维度继承周期性发生Watermark，不然在极度环境下会有很大的延时。

以是Watermark的天生方法必要按照营业场景的差异举办差异的选择。

Watermark的接口界说

对应Apache Flink Watermark两种差异的天生方法，我们相识一下对应的接口界说，如下：

• Periodic Watermarks - AssignerWithPeriodicWatermarks

/** 
* Returns the current watermark. This method is periodically called by the 
* system to retrieve the current watermark. The method may return {@code null} to 
* indicate that no new Watermark is available. 
* 
* <p>The returned watermark will be emitted only if it is non-null and itsTimestamp 
* is larger than that of the previously emitted watermark (to preserve the contract of 
* ascending watermarks). If the current watermark is still 
* identical to the previous one, no progress in EventTime has happened since 
* the previous call to this method. If a null value is returned, or theTimestamp 
* of the returned watermark is smaller than that of the last emitted one, then no 
* new watermark will be generated. 
* 
* <p>The interval in which this method is called and Watermarks are generated 
* depends on {@link ExecutionConfig#getAutoWatermarkInterval()}. 
* 
* @see org.Apache.flink.streaming.api.watermark.Watermark 
* @see ExecutionConfig#getAutoWatermarkInterval() 
* 
* @return {@code Null}, if no watermark should be emitted, or the next watermark to emit. 
*/ 
@Nullable 
Watermark getCurrentWatermark(); 

• Punctuated Watermarks - AssignerWithPunctuatedWatermarks

public interface AssignerWithPunctuatedWatermarks<T> extendsTimestampAssigner<T> { 
 
/** 
* Asks this implementation if it wants to emit a watermark. This method is called right after 
* the {@link #extractTimestamp(Object, long)} method. 
* 
* <p>The returned watermark will be emitted only if it is non-null and itsTimestamp 
* is larger than that of the previously emitted watermark (to preserve the contract of 
* ascending watermarks). If a null value is returned, or theTimestamp of the returned 
* watermark is smaller than that of the last emitted one, then no new watermark will 
* be generated. 
* 
* <p>For an example how to use this method, see the documentation of 
* {@link AssignerWithPunctuatedWatermarks this class}. 
* 
* @return {@code Null}, if no watermark should be emitted, or the next watermark to emit. 
*/ 
@Nullable 
Watermark checkAndGetNextWatermark(T lastElement, long extractedTimestamp);} 

• AssignerWithPunctuatedWatermarks 担任了TimestampAssigner接口 -TimestampAssigner

public interfaceTimestampAssigner<T> extends Function { 
 
/** 
* Assigns aTimestamp to an element, in milliseconds since the Epoch. 
* 
* <p>The method is passed the previously assignedTimestamp of the element. 
* That previousTimestamp may have been assigned from a previous assigner, 
* by ingestionTime. If the element did not carry aTimestamp before, this value is 
* {@code Long.MIN_VALUE}. 
* 
* @param element The element that theTimestamp is wil be assigned to. 
* @param previousElementTimestamp The previous internalTimestamp of the element, 
* or a negative value, if noTimestamp has been assigned, yet. 
* @return The newTimestamp. 
*/ 
long extractTimestamp(T element, long previousElementTimestamp); 
} 

从接口界说可以看出，Watermark可以在Event(Element)中提取EventTime，进而界说必然的计较逻辑发生Watermark的时刻戳。

Watermark办理如上题目

（编辑：湖南网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!