為何要保證流計算中的數據不丟失？

在領域，有些應用的對系統的可用性要求不那么苛刻，允許在系統異常時丟失一些數據。比如，實時推薦系統或業務質量實時監測系統因系統原因或應用程序故障而不能提供實時服務，在故障過程中流數據未進行處理就丟失了。這類場景對少量數據丟失是可容忍的，一是因為它們處理不是關鍵交易（不直接影響賬戶、交易、訂單等核心數據）；二是因為系統恢復后再處理“過時”數據意義已經不大，如客戶已經離開特定的商圈，系統還拿“過時”的位置信息向他推薦商品會讓客戶體驗很差。

然而，很多實時處理系統對數據的丟失是不容忍的，要求對所有數據至少處理一次或準確地處理一次。如CDR話單實時處理，實時計費等應用要求所有數據至少且至多處理一次，因為一旦出錯就會造成損失。

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

傳統的數據庫依賴日志、事務控制、Checkpoint等技術實現數據的ACID。而Streams流計算采用了“一致區域（Consistent Region）”機制實現流數據“至少處理一次”。

什么是一致區域

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

什么是一致狀態

在一致區域里的所有流的所有元組tuples全部經過一致區域里的operators完全處理的當前狀態。

下圖中的左邊是由3個Operator組成的應用，tuple經過op1處理后分布輸出到op2和op3，右邊表示tuple提交的時間軸。根據“一致狀態”的定義，第一條虛線所捕獲的狀態是“一致的”，即當前時刻的tuples都在3個Operators處理并提交。第二條和第三條虛線所捕獲的狀態是不一致的，因為在捕獲每個Operator狀態那個時間點，同一個tuple并未在一致區域內的完全處理完成。

一致區域的特點

• 一致區域的第一個operator必須能重發數據。
• 一致區域保證每一條數據至少被處理一次（at-least-once）。
• 一致區域里的所有operators要滿足如下條件之一，則保證每個元組僅被處理一次（Exactly Once）：
• 能夠將自身的狀態和交互的外部系統的狀態重置為最后一次檢查點一致狀態
• 能夠檢查重復的元組tuples，并且不予處理
• 不管處理幾次，都和第一次處理結果一樣
• 支持多線程的Operator。
• 允許用戶自定義一致區域的開始和結束（默認沒有輸出端口的Operator是一致區域的結束）。

實現原理

實現一致區域關鍵是定時建立該區域的一致狀態。從一致區域內的operator角度來看，建立一致狀態包含2個階段：(1)排干和(2)檢查點。當所有Operators完成這兩個階段，則一致區域的一致狀態被成功建立。

排干/Drain - 在這個階段，Operator排干它的內部狀態和輸出流。這意味著Operator已經完成處理之前收到的任何元組，排出其內部狀態并提交任何Pending的元組到其內部緩沖區輸出流。用通俗的話講，這個Operator完成排干后，對之前收到的任何元組來說它已經完全解放了。下游的Operator在排干完成之前，會完全處理前面Operator發送過來的元組，直到確定一致狀態。如果一個具有輸入流的Operator在一致區域，它的排干階段一定是發生在上游Operators排干完成之后和處理完所有上游排干時發過來的所有元組之后。

檢查點/checkpointing - 在這個階段,Operater序列化它的狀態到檢查點后端（后端可以是Redis或文件系統），這一階段總是發生排干階段之后。Streams runtime自動管理Operator的檢查點，這意味著runtime維護多個檢查點版本和自動刪除無需用于失敗恢復的檢查點。

恢復一致狀態

一旦發生異常，Streams runtime自動將一致區域恢復到最新的一致狀態。任何一個Operator的狀態重置一定是發生在其上游Operators已完成狀態重置之后。當一致區域內的全部Operator完成重置，這時就恢復到了一致狀態。之后就可以重放元組（最后確定一致狀態之后的的元組），這點類似數據庫的事務在rollback之后重新從第一條語句開始執行事務。

恢復一致狀態的操作在這幾個場景下發生：1）PE crash。2）主機crash。3）PE之間的連接中斷。4）Operator發現并報告錯誤。5）通過人工管理介入。

關鍵技術 - Chandy-Lamport算法

Chandy-Lamport算法是一個經典的分布式快照記錄算法。

為達到一致性狀態，并且減少建立一致狀態而對性能帶來的影響，Streams使用Chandy-Lamport算法的變體來建立一個分布式的快照。該算法會考慮流連接的渠道狀態channel state和運算符的進程狀態process state永久性存儲。Streams對一致性有著嚴格的定義：每一個提交的tuple必須被處理。因此，SPL系統并不需要保存渠道狀態。這是因為它有效地強制任何渠道狀態在進程狀態中得到映射。

如何使用？

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

可在允許的操作符上使用 @consistent 注釋定義一致區域的開頭。Streams自動確定一致區域的作用域，但是您可使用 @autonomous 注釋更改該區域的結束操作符。所定義的一致區域將定期建立一致狀態。您的應用程序還必須具有新的Job Control Plane操作符，該操作符將協調一致區域的排干 (draining) 和重置。具體語法如下：

@consistent( //一致區域定義

trigger={periodic|operatorDriven}, //如何觸發開始建立一個一致狀態

period=3.0, //周期

drainTimeout=30.0, //排干超時

resetTimeout=30.0, //重置超時

maxConsecutiveResetAttempts=5) //重置重試次數

詳情請咨詢！

客服熱線：023-66090381