流計(jì)算也能像數(shù)據(jù)庫(kù)那樣保證數(shù)據(jù)不丟失嗎？

---

為何要保證流計(jì)算中的數(shù)據(jù)不丟失？

在領(lǐng)域，有些應(yīng)用的對(duì)系統(tǒng)的可用性要求不那么苛刻，允許在系統(tǒng)異常時(shí)丟失一些數(shù)據(jù)。比如，實(shí)時(shí)推薦系統(tǒng)或業(yè)務(wù)質(zhì)量實(shí)時(shí)監(jiān)測(cè)系統(tǒng)因系統(tǒng)原因或應(yīng)用程序故障而不能提供實(shí)時(shí)服務(wù)，在故障過(guò)程中流數(shù)據(jù)未進(jìn)行處理就丟失了。這類(lèi)場(chǎng)景對(duì)少量數(shù)據(jù)丟失是可容忍的，一是因?yàn)樗鼈兲幚聿皇顷P(guān)鍵交易（不直接影響賬戶(hù)、交易、訂單等核心數(shù)據(jù)）；二是因?yàn)橄到y(tǒng)恢復(fù)后再處理“過(guò)時(shí)”數(shù)據(jù)意義已經(jīng)不大，如客戶(hù)已經(jīng)離開(kāi)特定的商圈，系統(tǒng)還拿“過(guò)時(shí)”的位置信息向他推薦商品會(huì)讓客戶(hù)體驗(yàn)很差。

然而，很多實(shí)時(shí)處理系統(tǒng)對(duì)數(shù)據(jù)的丟失是不容忍的，要求對(duì)所有數(shù)據(jù)至少處理一次或準(zhǔn)確地處理一次。如CDR話單實(shí)時(shí)處理，實(shí)時(shí)計(jì)費(fèi)等應(yīng)用要求所有數(shù)據(jù)至少且至多處理一次，因?yàn)橐坏┏鲥e(cuò)就會(huì)造成損失。

如何做到“至少處理一次”？

傳統(tǒng)的數(shù)據(jù)庫(kù)依賴(lài)日志、事務(wù)控制、Checkpoint等技術(shù)實(shí)現(xiàn)數(shù)據(jù)的ACID。而Streams流計(jì)算采用了“一致區(qū)域（Consistent Region）”機(jī)制實(shí)現(xiàn)流數(shù)據(jù)“至少處理一次”。

什么是一致區(qū)域

在應(yīng)用程序中定義的sub-graph（或叫Region），一旦有tuple流入該區(qū)域就要保證該tuple經(jīng)過(guò)了所有Operator的至少一次處理，那么該區(qū)域就是“一致區(qū)域”。

什么是一致?tīng)顟B(tài)

在一致區(qū)域里的所有流的所有元組tuples全部經(jīng)過(guò)一致區(qū)域里的operators完全處理的當(dāng)前狀態(tài)。

下圖中的左邊是由3個(gè)Operator組成的應(yīng)用，tuple經(jīng)過(guò)op1處理后分布輸出到op2和op3，右邊表示tuple提交的時(shí)間軸。根據(jù)“一致?tīng)顟B(tài)”的定義，第一條虛線所捕獲的狀態(tài)是“一致的”，即當(dāng)前時(shí)刻的tuples都在3個(gè)Operators處理并提交。第二條和第三條虛線所捕獲的狀態(tài)是不一致的，因?yàn)樵诓东@每個(gè)Operator狀態(tài)那個(gè)時(shí)間點(diǎn)，同一個(gè)tuple并未在一致區(qū)域內(nèi)的完全處理完成。

一致區(qū)域的特點(diǎn)

• 一致區(qū)域的第一個(gè)operator必須能重發(fā)數(shù)據(jù)。
• 一致區(qū)域保證每一條數(shù)據(jù)至少被處理一次（at-least-once）。
• 一致區(qū)域里的所有operators要滿(mǎn)足如下條件之一，則保證每個(gè)元組僅被處理一次（Exactly Once）：
• 能夠?qū)⒆陨淼臓顟B(tài)和交互的外部系統(tǒng)的狀態(tài)重置為最后一次檢查點(diǎn)一致?tīng)顟B(tài)
• 能夠檢查重復(fù)的元組tuples，并且不予處理
• 不管處理幾次，都和第一次處理結(jié)果一樣
• 支持多線程的Operator。
• 允許用戶(hù)自定義一致區(qū)域的開(kāi)始和結(jié)束（默認(rèn)沒(méi)有輸出端口的Operator是一致區(qū)域的結(jié)束）。

實(shí)現(xiàn)原理

實(shí)現(xiàn)一致區(qū)域關(guān)鍵是定時(shí)建立該區(qū)域的一致?tīng)顟B(tài)。從一致區(qū)域內(nèi)的operator角度來(lái)看，建立一致?tīng)顟B(tài)包含2個(gè)階段：(1)排干和(2)檢查點(diǎn)。當(dāng)所有Operators完成這兩個(gè)階段，則一致區(qū)域的一致?tīng)顟B(tài)被成功建立。

排干/Drain - 在這個(gè)階段，Operator排干它的內(nèi)部狀態(tài)和輸出流。這意味著Operator已經(jīng)完成處理之前收到的任何元組，排出其內(nèi)部狀態(tài)并提交任何Pending的元組到其內(nèi)部緩沖區(qū)輸出流。用通俗的話講，這個(gè)Operator完成排干后，對(duì)之前收到的任何元組來(lái)說(shuō)它已經(jīng)完全解放了。下游的Operator在排干完成之前，會(huì)完全處理前面Operator發(fā)送過(guò)來(lái)的元組，直到確定一致?tīng)顟B(tài)。如果一個(gè)具有輸入流的Operator在一致區(qū)域，它的排干階段一定是發(fā)生在上游Operators排干完成之后和處理完所有上游排干時(shí)發(fā)過(guò)來(lái)的所有元組之后。

檢查點(diǎn)/checkpointing - 在這個(gè)階段,Operater序列化它的狀態(tài)到檢查點(diǎn)后端（后端可以是Redis或文件系統(tǒng)），這一階段總是發(fā)生排干階段之后。Streams runtime自動(dòng)管理Operator的檢查點(diǎn)，這意味著runtime維護(hù)多個(gè)檢查點(diǎn)版本和自動(dòng)刪除無(wú)需用于失敗恢復(fù)的檢查點(diǎn)。

恢復(fù)一致?tīng)顟B(tài)

一旦發(fā)生異常，Streams runtime自動(dòng)將一致區(qū)域恢復(fù)到最新的一致?tīng)顟B(tài)。任何一個(gè)Operator的狀態(tài)重置一定是發(fā)生在其上游Operators已完成狀態(tài)重置之后。當(dāng)一致區(qū)域內(nèi)的全部Operator完成重置，這時(shí)就恢復(fù)到了一致?tīng)顟B(tài)。之后就可以重放元組（最后確定一致?tīng)顟B(tài)之后的的元組），這點(diǎn)類(lèi)似數(shù)據(jù)庫(kù)的事務(wù)在rollback之后重新從第一條語(yǔ)句開(kāi)始執(zhí)行事務(wù)。

恢復(fù)一致?tīng)顟B(tài)的操作在這幾個(gè)場(chǎng)景下發(fā)生：1）PE crash。2）主機(jī)crash。3）PE之間的連接中斷。4）Operator發(fā)現(xiàn)并報(bào)告錯(cuò)誤。5）通過(guò)人工管理介入。

關(guān)鍵技術(shù) - Chandy-Lamport算法

Chandy-Lamport算法是一個(gè)經(jīng)典的分布式快照記錄算法。

為達(dá)到一致性狀態(tài)，并且減少建立一致?tīng)顟B(tài)而對(duì)性能帶來(lái)的影響，Streams使用Chandy-Lamport算法的變體來(lái)建立一個(gè)分布式的快照。該算法會(huì)考慮流連接的渠道狀態(tài)channel state和運(yùn)算符的進(jìn)程狀態(tài)process state永久性存儲(chǔ)。Streams對(duì)一致性有著嚴(yán)格的定義：每一個(gè)提交的tuple必須被處理。因此，SPL系統(tǒng)并不需要保存渠道狀態(tài)。這是因?yàn)樗行У貜?qiáng)制任何渠道狀態(tài)在進(jìn)程狀態(tài)中得到映射。

如何使用？

Streams通過(guò)操作符Operator和注釋annotations得到增強(qiáng)，這些操作程序和注釋允許定義在流處理期間不會(huì)丟失元組的區(qū)域，確保元組在一致區(qū)域至少被處理一次。

可在允許的操作符上使用 @consistent 注釋定義一致區(qū)域的開(kāi)頭。Streams自動(dòng)確定一致區(qū)域的作用域，但是您可使用 @autonomous 注釋更改該區(qū)域的結(jié)束操作符。所定義的一致區(qū)域?qū)⒍ㄆ诮⒁恢聽(tīng)顟B(tài)。您的應(yīng)用程序還必須具有新的Job Control Plane操作符，該操作符將協(xié)調(diào)一致區(qū)域的排干 (draining) 和重置。具體語(yǔ)法如下：

@consistent( //一致區(qū)域定義

trigger={periodic|operatorDriven}, //如何觸發(fā)開(kāi)始建立一個(gè)一致?tīng)顟B(tài)

period=3.0, //周期

drainTimeout=30.0, //排干超時(shí)

resetTimeout=30.0, //重置超時(shí)

maxConsecutiveResetAttempts=5) //重置重試次數(shù)

詳情請(qǐng)咨詢(xún)！

客服熱線：023-66090381