1.亂序問(wèn)題

流處理從事件產(chǎn)生，到流經(jīng)source，再到operator，中間是有一個(gè)過(guò)程和時(shí)間的，雖然大部分情況下，流到operator的數(shù)據(jù)都是按照事件產(chǎn)生的時(shí)間順序來(lái)的，但是也不排除由于網(wǎng)絡(luò)、分布式等原因，導(dǎo)致亂序的產(chǎn)生，所謂亂序，就是指Flink接收到的事件的先后順序不是嚴(yán)格按照事件的Event Time順序排列的。

一旦出現(xiàn)亂序，如果只根據(jù)eventTime決定window的運(yùn)行，我們不能明確數(shù)據(jù)是否全部到位，但又不能無(wú)限期的等下去，此時(shí)必須要有個(gè)機(jī)制來(lái)保證一個(gè)特定的時(shí)間后，必須觸發(fā)window去進(jìn)行計(jì)算了，這個(gè)特別的機(jī)制，就是Watermark。

例如：某App會(huì)記錄用戶(hù)的所有點(diǎn)擊行為，并回傳日志（在網(wǎng)絡(luò)不好的情況下，先保存在本地，延后回傳）。A用戶(hù)在11:02對(duì)App進(jìn)行操作，B用戶(hù)在11:03對(duì)App進(jìn)行操作，但是A用戶(hù)的網(wǎng)絡(luò)不太穩(wěn)定，回傳日志延遲了，導(dǎo)致我們?cè)诜?wù)端先接受到B用戶(hù)11:03的消息，然后再接受到A用戶(hù)11:02的消息，消息亂序了。

2.什么是Watermark

Watermark是Apache Flink提出的一種用來(lái)解決亂序、延遲數(shù)據(jù)等情況的解決方案。

它是建立在事件時(shí)間上的一個(gè)概念，用來(lái)刻畫(huà)數(shù)據(jù)流的完整性。如果按照處理時(shí)間來(lái)衡量事件，一切都是有序的、完美的，自然而然也就不需要Watermark了。換句話說(shuō)事件時(shí)間帶來(lái)了亂序的問(wèn)題，而Watermark就是用來(lái)解決亂序問(wèn)題。所謂的亂序，其實(shí)就是有事件延遲了，對(duì)于延遲的元素，我們不可能無(wú)限期的等下去，必須要有一種機(jī)制來(lái)保證一個(gè)特定的時(shí)間后，必須觸發(fā)Window進(jìn)行計(jì)算。這個(gè)特別的機(jī)制,就是Watermark,它告訴了算子延遲到達(dá)的消息不應(yīng)該再被接收。

watermrk具體特點(diǎn)如下：

• Watermark是一種衡量Event Time進(jìn)展的機(jī)制。
• Watermark是用于處理亂序事件的，通常用Watermark機(jī)制結(jié)合window來(lái)實(shí)現(xiàn)。
• 數(shù)據(jù)流中的Watermark用于表示timestamp小于Watermark的數(shù)據(jù)，都已經(jīng)到達(dá)了，因此，window的執(zhí)行也是由Watermark觸發(fā)的。
• Watermark可以理解成一個(gè)延遲觸發(fā)機(jī)制，我們可以設(shè)置Watermark的延時(shí)時(shí)長(zhǎng)t，每次系統(tǒng)會(huì)校驗(yàn)已經(jīng)到達(dá)的數(shù)據(jù)中最大的maxEventTime，然后認(rèn)定eventTime小于maxEventTime - t的所有數(shù)據(jù)都已經(jīng)到達(dá)，如果有窗口的停止時(shí)間等于maxEventTime – t，那么這個(gè)窗口被觸發(fā)執(zhí)行。
• watermark 用來(lái)讓程序自己平衡延遲和結(jié)果正確性。

3.Watermark原理

Watermark會(huì)攜帶一個(gè)單調(diào)遞增的時(shí)間戳t，Watermark(t)表示所有時(shí)間戳不大于t的數(shù)據(jù)都已經(jīng)到來(lái)了，未來(lái)小于等于t的數(shù)據(jù)不會(huì)再來(lái)，因此可以放心地觸發(fā)和銷(xiāo)毀窗口了。

當(dāng)Flink,接收到數(shù)據(jù)時(shí)，會(huì)按照一定的規(guī)則去生成Watermark，這條Watermark就等于當(dāng)前所有到達(dá)數(shù)據(jù)中的maExertT me"-延N時(shí)長(zhǎng)，也就定說(shuō)，Watermark是基于數(shù)據(jù)攜帶的時(shí)間戳生成的，一旦Watermark比當(dāng)前未觸發(fā)的窗口的停止時(shí)間要晚，那么就會(huì)觸發(fā)相應(yīng)窗口的執(zhí)行。由于eventtime是由數(shù)據(jù)攜帶的，因此，如果運(yùn)行過(guò)程中無(wú)法獲取新的數(shù)據(jù)，那么沒(méi)有被觸發(fā)的窗口將永遠(yuǎn)都不被觸發(fā)。

4.延遲數(shù)據(jù)處理機(jī)制

Watermark能夠應(yīng)對(duì)亂序的數(shù)據(jù)，但是真實(shí)世界中沒(méi)法得到一個(gè)完美的 Watermark數(shù)值。要么沒(méi)法獲取到，要么耗費(fèi)太大，因此實(shí)際工作中會(huì)近似 Watermark(t)之后，還有較小的概率接收到時(shí)間戳t之前的數(shù)據(jù)，在Flink中將這些數(shù)據(jù)定義為“l(fā)ate elements”，可以在Window中指定允許延遲的最大時(shí)間（默認(rèn)為О)。

延遲事件是亂序事件的特例，和一般亂序事件不同的是它們的亂序程度超出了水位線( Watermark)的預(yù)計(jì)，導(dǎo)致窗口在它們到達(dá)之前已經(jīng)關(guān)閉。

延遲事件出現(xiàn)時(shí)窗口已經(jīng)關(guān)閉并產(chǎn)出了計(jì)算結(jié)果，對(duì)于此種情況處理的方法有3種:

• 重新激活已經(jīng)關(guān)閉的窗口并重新計(jì)算以修正結(jié)果。
• 將延遲事件收集起來(lái)另外處理。
• 將延遲事件視為錯(cuò)誤消息并丟棄。

Flink默認(rèn)的處理方式是第3種直接丟棄，其他兩種方式分別使用Side Output和AllowedLateness。

• Side Output機(jī)制：將延遲事件單獨(dú)放入一個(gè)數(shù)據(jù)流分支，這會(huì)作為Window計(jì)算結(jié)果的副產(chǎn)品，以便用戶(hù)獲取并對(duì)其進(jìn)行特殊處理。遲來(lái)的數(shù)據(jù)同樣可以觸發(fā)窗口，進(jìn)行輸出。
• Allowed Lateness機(jī)制：允許用戶(hù)設(shè)置一個(gè)允許的最大延遲時(shí)長(zhǎng)。Flink會(huì)在窗口關(guān)閉后一直保存窗口的狀態(tài)直至超過(guò)允許延遲時(shí)長(zhǎng)，這期間的延遲事件不會(huì)被丟棄，而是默認(rèn)會(huì)觸發(fā)窗口重新計(jì)算。因?yàn)楸４娲翱跔顟B(tài)需要額外內(nèi)存，并且如果窗口計(jì)算使用了ProcessWindowFunction APl還可能使得每個(gè)延遲事件觸發(fā)一次窗口的全量計(jì)算，代價(jià)比較大，所以允許延遲時(shí)長(zhǎng)不宜設(shè)得太長(zhǎng)，延遲事件也不宜過(guò)多。