對(duì)于流處理系統(tǒng)，實(shí)現(xiàn)exactly-once非常重要，否則系統(tǒng)輸出的結(jié)果便會(huì)有重復(fù)或者遺漏數(shù)據(jù)。在流處理系統(tǒng)剛興起的時(shí)候，結(jié)果不正確的情況是默認(rèn)允許的，人們?cè)诋?dāng)時(shí)提出了lambda架構(gòu)來解決這個(gè)問題，lambda架構(gòu)包含兩套系統(tǒng)，一套是結(jié)果不怎么準(zhǔn)確的流系統(tǒng)，另一套是結(jié)果精確的批處理系統(tǒng)，流系統(tǒng)可以實(shí)時(shí)給出報(bào)表數(shù)據(jù)，但是這個(gè)數(shù)據(jù)不怎么準(zhǔn)確，需要定時(shí)的用批處理系統(tǒng)的結(jié)果去修正。lambda架構(gòu)帶來了一系列問題:

（1）不準(zhǔn)確性：有時(shí)候會(huì)低估這種不準(zhǔn)確性。

（2）不一致性：已經(jīng)給客戶展示的數(shù)據(jù)有可能變化

（3）復(fù)雜性：同時(shí)維護(hù)兩套系統(tǒng)比較復(fù)雜

（4）不可預(yù)測(cè)性

（5）延遲比較大

Beam(Data flow)端到端一致性的實(shí)現(xiàn)

端到端的exactly-once需要考慮三個(gè)中間步驟

（1）從源去讀

（2）算子的shuffle

（3）寫到sink

shuffle的時(shí)候保證exactly once

Dataflow通過rpc來完成shuffle，在rpc移動(dòng)數(shù)據(jù)的時(shí)候，如果rpc返回的結(jié)果不是成功，會(huì)一直重試，直到返回成功，這樣的情況下數(shù)據(jù)雖然不會(huì)遺失，但是會(huì)出現(xiàn)重復(fù)，那么如何避免重復(fù)呢？

Dataflow給每一條要處理的數(shù)據(jù)加上了唯一標(biāo)識(shí)ID，同時(shí)shuffle的接受方會(huì)對(duì)ID進(jìn)行判重處理，如果發(fā)現(xiàn)ID已經(jīng)處理過，會(huì)丟掉當(dāng)前的這條數(shù)據(jù)。Dataflow利用kv存儲(chǔ)作為底層存儲(chǔ)實(shí)現(xiàn)判重的功能。

解決不可重復(fù)操作

處理過程的某些算子操作是不可重復(fù)的，也就是說數(shù)據(jù)流每次運(yùn)行的結(jié)果是不一樣的，這種情況下多次重試去執(zhí)行數(shù)據(jù)流，并將結(jié)果通過RPC發(fā)送到對(duì)方是不正確的。Dataflow通過checkpoint技術(shù)來解決這個(gè)問題。簡(jiǎn)單說通過checkpoint將每個(gè)算子的輸出保存到存儲(chǔ)上，在shuffle的時(shí)候，RPC發(fā)送的數(shù)據(jù)是從存儲(chǔ)去讀，這樣就避免執(zhí)行之前的算子，這樣保證重試傳輸?shù)臅r(shí)候發(fā)送的數(shù)據(jù)還是以前產(chǎn)生好的數(shù)據(jù)。

性能優(yōu)化

前邊提到的exactly-once實(shí)現(xiàn)起來會(huì)非常的低效，為了解決性能問題，Dataflow實(shí)現(xiàn)了幾個(gè)優(yōu)化措施：計(jì)算圖優(yōu)化和布隆過濾器

1. 計(jì)算圖優(yōu)化

（1）多個(gè)邏輯算子合并成一個(gè)物理計(jì)算單元，這樣不需要存儲(chǔ)每個(gè)算子的輸出，而是為整個(gè)計(jì)算單元存儲(chǔ)輸出

（2）在shuffle前進(jìn)行預(yù)聚合，這樣可以減少shuffle的數(shù)據(jù)量

2. 布隆過濾器

前邊提到的去重操作依賴于kv存儲(chǔ)，而存儲(chǔ)的讀寫比較耗時(shí)。根據(jù)布隆過濾器的false positive特性，以及重復(fù)數(shù)據(jù)比較少的情況，我們通過增加布隆過濾器來提高重復(fù)判斷。接收方在收到接受的數(shù)據(jù)記錄，首先通過布隆過濾器來判斷ID是否存在，如果不存在，那么一定不存在，我們就不需要再去讀存儲(chǔ)去判斷。如果在布隆過濾器中存在，那么需要進(jìn)一步讀存儲(chǔ)來判斷ID是否真的存在。這樣我們就減少了讀存儲(chǔ)的次數(shù)，從而提升性能。

當(dāng)在布隆過濾器存儲(chǔ)的數(shù)據(jù)比較多的時(shí)候，返回false的概率就會(huì)降低，為了避免這種情況，Dataflow會(huì)定時(shí)產(chǎn)生一個(gè)新的布隆過濾器，而系統(tǒng)的每條數(shù)據(jù)附帶著timestamp, 可以通過timestamp來選取對(duì)應(yīng)的布隆過濾器

3. 垃圾回收

為了去重，Dataflow需要存儲(chǔ)一系列的id在存儲(chǔ)上，但是隨著時(shí)間推移，數(shù)據(jù)會(huì)越來越多，怎么去做數(shù)據(jù)清理工作呢？Dataflow通過產(chǎn)生垃圾回收watermark來決定清楚已經(jīng)過期的數(shù)據(jù)。