Apache Flink 漫談系列 - Watermark是個(gè)啥？

作者：孫金城 2022-07-13 13:03:29

本文以一個(gè)流計(jì)算常見的亂序問題介紹了Apache Flink如何利用Watermark機(jī)制來處理亂序問題。

實(shí)際問題（亂序）

在介紹Watermark相關(guān)內(nèi)容之前我們先拋出一個(gè)具體的問題，在實(shí)際的流式計(jì)算中數(shù)據(jù)到來的順序?qū)τ?jì)算結(jié)果的正確性有至關(guān)重要的影響，比如：某數(shù)據(jù)源中的某些數(shù)據(jù)由于某種原因(如：網(wǎng)絡(luò)原因，外部存儲(chǔ)自身原因)會(huì)有5秒的延時(shí)，也就是在實(shí)際時(shí)間的第1秒產(chǎn)生的數(shù)據(jù)有可能在第5秒中產(chǎn)生的數(shù)據(jù)之后到來(比如到Window處理節(jié)點(diǎn)).選具體某個(gè)delay的元素來說，假設(shè)在一個(gè)5秒的Tumble窗口(詳見后續(xù)Window篇介紹)，有一個(gè)EventTime是 11秒的數(shù)據(jù)，在第16秒時(shí)候到來了。圖示第11秒的數(shù)據(jù)，在16秒到來了，如下圖：

那么對(duì)于一個(gè)Count聚合的Tumble(5s)的window，上面的情況如何處理才能window2=4，window3=2 呢？

Apache Flink的時(shí)間類型

開篇我們描述的問題是一個(gè)很常見的TimeWindow中數(shù)據(jù)亂序的問題，亂序是相對(duì)于事件產(chǎn)生時(shí)間和到達(dá)Apache Flink 實(shí)際處理算子的順序而言的，關(guān)于時(shí)間在Apache Flink中有如下三種時(shí)間類型，如下圖：

ProcessingTime 是數(shù)據(jù)流入到具體某個(gè)算子時(shí)候相應(yīng)的系統(tǒng)時(shí)間。ProcessingTime 有最好的性能和最低的延遲。但在分布式計(jì)算環(huán)境中ProcessingTime具有不確定性，相同數(shù)據(jù)流多次運(yùn)行有可能產(chǎn)生不同的計(jì)算結(jié)果。
IngestionTimeIngestionTime是數(shù)據(jù)進(jìn)入Apache Flink框架的時(shí)間，是在Source Operator中設(shè)置的。與ProcessingTime相比可以提供更可預(yù)測(cè)的結(jié)果，因?yàn)镮ngestionTime的時(shí)間戳比較穩(wěn)定(在源處只記錄一次)，同一數(shù)據(jù)在流經(jīng)不同窗口操作時(shí)將使用相同的時(shí)間戳，而對(duì)于ProcessingTime同一數(shù)據(jù)在流經(jīng)不同窗口算子會(huì)有不同的處理時(shí)間戳。
EventTimeEventTime是事件在設(shè)備上產(chǎn)生時(shí)候攜帶的。在進(jìn)入Apache Flink框架之前EventTime通常要嵌入到記錄中，并且EventTime也可以從記錄中提取出來。在實(shí)際的網(wǎng)上購物訂單等業(yè)務(wù)場(chǎng)景中，大多會(huì)使用EventTime來進(jìn)行數(shù)據(jù)計(jì)算。

開篇描述的問題和本篇要介紹的Watermark所涉及的時(shí)間類型均是指EventTime類型。

什么是Watermark

Watermark是Apache Flink為了處理EventTime 窗口計(jì)算提出的一種機(jī)制,本質(zhì)上也是一種時(shí)間戳，由Apache Flink Source或者自定義的Watermark生成器按照需求Punctuated或者Periodic兩種方式生成的一種系統(tǒng)Event，與普通數(shù)據(jù)流Event一樣流轉(zhuǎn)到對(duì)應(yīng)的下游算子，接收到Watermark Event的算子以此不斷調(diào)整自己管理的EventTime clock。Apache Flink 框架保證Watermark單調(diào)遞增，算子接收到一個(gè)Watermark時(shí)候，框架知道不會(huì)再有任何小于該Watermark的時(shí)間戳的數(shù)據(jù)元素到來了，所以Watermark可以看做是告訴Apache Flink框架數(shù)據(jù)流已經(jīng)處理到什么位置(時(shí)間維度)的方式。Watermark的產(chǎn)生和Apache Flink內(nèi)部處理邏輯如下圖所示:

Watermark的產(chǎn)生方式

目前Apache Flink 有兩種生產(chǎn)Watermark的方式，如下：

Punctuated數(shù)據(jù)流中每一個(gè)遞增的EventTime都會(huì)產(chǎn)生一個(gè)Watermark。在實(shí)際的生產(chǎn)中Punctuated方式在TPS很高的場(chǎng)景下會(huì)產(chǎn)生大量的Watermark在一定程度上對(duì)下游算子造成壓力，所以只有在實(shí)時(shí)性要求非常高的場(chǎng)景才會(huì)選擇Punctuated的方式進(jìn)行Watermark的生成。
Periodic周期性的（一定時(shí)間間隔或者達(dá)到一定的記錄條數(shù)）產(chǎn)生一個(gè)Watermark。在實(shí)際的生產(chǎn)中Periodic的方式必須結(jié)合時(shí)間和積累條數(shù)兩個(gè)維度繼續(xù)周期性產(chǎn)生Watermark，否則在極端情況下會(huì)有很大的延時(shí)。

所以Watermark的生成方式需要根據(jù)業(yè)務(wù)場(chǎng)景的不同進(jìn)行不同的選擇。

Watermark的接口定義

對(duì)應(yīng)Apache Flink Watermark兩種不同的生成方式，我們了解一下對(duì)應(yīng)的接口定義，如下：

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> extends TimestampAssigner<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 its timestamp
   * is larger than that of the previously emitted watermark (to preserve the contract of
   * ascending watermarks). If a null value is returned, or the timestamp 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 interface TimestampAssigner<T> extends Function {

  /**
   * Assigns a timestamp to an element, in milliseconds since the Epoch.
   *
   * <p>The method is passed the previously assigned timestamp of the element.
   * That previous timestamp may have been assigned from a previous assigner,
   * by ingestion time. If the element did not carry a timestamp before, this value is
   * {@code Long.MIN_VALUE}.
   *
   * @param element The element that the timestamp will be assigned to.
   * @param previousElementTimestamp The previous internal timestamp of the element,
   *                                 or a negative value, if no timestamp has been assigned yet.
   * @return The new timestamp.
   */
  long extractTimestamp(T element, long previousElementTimestamp);
}

從接口定義可以看出，Watermark可以在Event(Element)中提取EventTime，進(jìn)而定義一定的計(jì)算邏輯產(chǎn)生Watermark的時(shí)間戳。

Watermark解決如上問題

從上面的Watermark生成接口和Apache Flink內(nèi)部對(duì)Periodic Watermark的實(shí)現(xiàn)來看，Watermark的時(shí)間戳可以和Event中的EventTime 一致，也可以自己定義任何合理的邏輯使得Watermark的時(shí)間戳不等于Event中的EventTime，Event中的EventTime自產(chǎn)生那一刻起就不可以改變了，不受Apache Flink框架控制，而Watermark的產(chǎn)生是在Apache Flink的Source節(jié)點(diǎn)或?qū)崿F(xiàn)的Watermark生成器計(jì)算產(chǎn)生(如上Apache Flink內(nèi)置的 Periodic Watermark實(shí)現(xiàn)), Apache Flink內(nèi)部對(duì)單流或多流的場(chǎng)景有統(tǒng)一的Watermark處理。

回過頭來我們?cè)诳纯碬atermark機(jī)制如何解決上面的問題，上面的問題在于如何將遲來的EventTime 位11的元素正確處理。要解決這個(gè)問題我們還需要先了解一下EventTime window是如何觸發(fā)的？EventTime window 計(jì)算條件是當(dāng)Window計(jì)算的Timer時(shí)間戳小于等于當(dāng)前系統(tǒng)的Watermak的時(shí)間戳?xí)r候進(jìn)行計(jì)算。

當(dāng)Watermark的時(shí)間戳等于Event中攜帶的EventTime時(shí)候，上面場(chǎng)景（Watermark=EventTime)的計(jì)算結(jié)果如下：

上面對(duì)應(yīng)的DDL定義如下：

create table t1(
  ts timestamp(3),
  other bigint,
  WATERMARK FOR ts AS ts
) with (
  'connector' = 'xx'
)

如果想正確處理遲來的數(shù)據(jù)可以定義Watermark生成策略為 Watermark = EventTime -5s，如下：

上面對(duì)應(yīng)的DDL定義如下：

create table t1(
  ts timestamp(3),
  other bigint,
  WATERMARK FOR ts AS ts - interval '5' SECOND
) with (
  'connector' = 'xx'
)

上面正確處理的根源是我們采取了延遲觸發(fā) window 計(jì)算的方式正確處理了 Late Event. 與此同時(shí)，我們發(fā)現(xiàn)window的延時(shí)觸發(fā)計(jì)算，也導(dǎo)致了下游的LATENCY變大，本例子中下游得到window的結(jié)果就延遲了5s.

多流的Watermark處理

在實(shí)際的流計(jì)算中往往一個(gè)job中會(huì)處理多個(gè)Source的數(shù)據(jù)，對(duì)Source的數(shù)據(jù)進(jìn)行GroupBy分組，那么來自不同Source的相同key值會(huì)shuffle到同一個(gè)處理節(jié)點(diǎn)，并攜帶各自的Watermark，Apache Flink內(nèi)部要保證Watermark要保持單調(diào)遞增，多個(gè)Source的Watermark匯聚到一起時(shí)候可能不是單調(diào)自增的，這樣的情況Apache Flink內(nèi)部是如何處理的呢？如下圖所示：

Apache Flink內(nèi)部實(shí)現(xiàn)每一個(gè)邊上只能有一個(gè)遞增的Watermark，當(dāng)出現(xiàn)多流攜帶Eventtime匯聚到一起(Join or Union)時(shí)候，Apache Flink會(huì)選擇所有流入的Eventtime中最小min(stream1, stream2...streamN)的一個(gè)向下游流出。從而保證watermark的單調(diào)遞增和保證數(shù)據(jù)的完整性.如下圖:

小結(jié)

本節(jié)以一個(gè)流計(jì)算常見的亂序問題介紹了Apache Flink如何利用Watermark機(jī)制來處理亂序問題. 本篇內(nèi)容在一定程度上也體現(xiàn)了EventTime Window中的Trigger機(jī)制依賴了Watermark(后續(xù)Window篇章會(huì)介紹)。Watermark機(jī)制是流計(jì)算中處理亂序，正確處理Late Event的核心手段。更多細(xì)節(jié)歡迎關(guān)注《Apache Flink 知其然，知其所以然》系列視頻課程！

作者介紹

孫金城，51CTO社區(qū)編輯，Apache Flink PMC 成員，Apache Beam Committer，Apache IoTDB PMC 成員，ALC Beijing 成員，Apache ShenYu 導(dǎo)師，Apache 軟件基金會(huì)成員。關(guān)注技術(shù)領(lǐng)域流計(jì)算和時(shí)序數(shù)據(jù)存儲(chǔ)。

責(zé)任編輯：張燕妮來源：孫金城

流計(jì)算亂序