對(duì)于流處理引擎來說，處理延遲到達(dá)的事件是至關(guān)重要的功能。 解決這個(gè)問題的方法是加水位線的概念。 從Spark 2.1開始，結(jié)構(gòu)化流API就支持它。

什么是水位線？

加水位線是一種有用的方法，可幫助流處理引擎處理延遲。 基本上，水印是一個(gè)閾值，用于指定系統(tǒng)等待延遲事件的時(shí)間。 如果到達(dá)事件位于水位線之內(nèi)，它將用于更新查詢。 否則，如果它早于水位線，它將被丟棄，并且流引擎不會(huì)對(duì)其進(jìn)行進(jìn)一步處理。

> Flooding watermarks

如何使用它？

自Spark 2.1起，水位線被引入到結(jié)構(gòu)化流API中。 您可以通過將withWatermark-Operator添加到查詢中來啟用它：

withWatermark（eventTime：String，delayThreshold：String）：數(shù)據(jù)集[T]

它需要兩個(gè)參數(shù)，a）一個(gè)事件時(shí)間列（必須與聚合正在處理的列相同）和b）一個(gè)閾值，用于指定應(yīng)處理多長時(shí)間的延遲數(shù)據(jù)（以事件時(shí)間為單位）。 然后，Spark將維持聚合狀態(tài)，直到max eventTime — delayThreshold> T，其中max eventTime是引擎看到的最新事件時(shí)間，T是窗口的開始時(shí)間。 如果后期數(shù)據(jù)落入此閾值之內(nèi)，則查詢將最終得到更新（下圖中的右圖）。 否則，它將被丟棄，并且不會(huì)觸發(fā)任何重新處理（下圖中的左圖）。

> Late donkey in structured word count: event dropped (left), event within watermark updates Window

值得一提的是，查詢的輸出模式必須設(shè)置為"追加"（默認(rèn)）或"更新"。完全模式不能與設(shè)計(jì)中的水印結(jié)合使用，因?yàn)樗枰?要保存的數(shù)據(jù)，用于將整個(gè)結(jié)果表輸出到接收器。

可以在這里找到如何在簡單的Spark結(jié)構(gòu)化流應(yīng)用程序中使用該概念的快速演示-它是字?jǐn)?shù)統(tǒng)計(jì)（對(duì)NLP進(jìn)行了一些小的增強(qiáng)），還有其他：D

但是，為什么我要關(guān)心？

在分布式和聯(lián)網(wǎng)的系統(tǒng)中，總會(huì)有中斷的機(jī)會(huì)-節(jié)點(diǎn)故障，傳感器丟失連接等等。 因此，不能保證數(shù)據(jù)將按創(chuàng)建順序到達(dá)流處理引擎。 為了容錯(cuò)，因此有必要處理此類亂序數(shù)據(jù)。

為了解決此問題，必須保留聚合狀態(tài)。 如果發(fā)生延遲事件，則可以重新處理查詢。 但這意味著所有聚合的狀態(tài)必須無限期地保持，這也導(dǎo)致內(nèi)存使用量也無限期地增長。 除非系統(tǒng)具有無限的資源（即無限的預(yù)算），否則在現(xiàn)實(shí)世界中這是不切實(shí)際的。 因此，加水位線是一個(gè)有用的概念，可以通過設(shè)計(jì)約束系統(tǒng)并防止其在運(yùn)行時(shí)爆炸。