1.5 一致性

讓批處理數據處理程序在由廉價機器組成的集群上可靠地運行不是一件容易的事情，流處理程序則更難。可靠運行的核心問題是如何保證分布式系統有狀態計算的一致性。本節將分析在Flink架構中容錯與一致性的實現方案，即異步屏障快照技術。

1.5.1 有狀態計算

在Example1.1中，聚合結果與本窗口內的所有記錄有關，由于每個記錄都是一個獨立事件，窗口需要緩存這些獨立事件或由這些獨立事件產生的中間結果。這類聚合運算被稱為有狀態計算，而基于單個事件的過濾處理則被稱為無狀態計算，如圖1-33所示。

Flink有以下兩類狀態。

（1）數據處理應用程序自定義的狀態，這類狀態由應用程序創建維護。

（2）引擎定義的狀態，這類狀態由引擎負責管理，如窗口緩存的事件及中間聚合結果。

1.5.2 exactly-once語義

在分布式系統中，所有數據備份在同一時刻的值是相同的，或者說所有客戶端讀取的值是相同的，這就是一致性的含義。根據正確性級別的不同，一致性有以下三種形式。

（1）at-most-once：盡可能正確，但不保證一定正確。對應Example1.1，在系統發生故障恢復后，聚合結果可能會出錯。

（2）at-least-once：對應Example1.1，在系統發生故障恢復后，聚合計算不會漏掉故障恢復之前窗口內的事件，但可能會重復計算某些事件，這通常用于實時性較高但準確性要求不高的場合。例如，Lambda架構將強實時性的Storm和強一致性的 Hadoop 批處理系統融合在一起，Storm 負責實時生成近似結果，Hadoop負責計算最終精準結果。

（3）exactly-once：對應Example1.1，在系統發生故障恢復后，聚合結果與假定沒有發生故障情況時一致。這種語義加大了高吞吐和低延遲的實現難度，異步屏障快照技術是Flink提供這種語義的理論基礎。

1.5.3 異步屏障快照

為了更好地理解異步屏障快照（ABS，Asynchronous Barrier Snapshot）理論，我們首先介紹幾個相關概念。

（1）檢查點（Checkpoint）：關系型數據庫并不會立即將提交的事物寫回磁盤，而是先寫入緩存（Buffer Cache）和重做日志（Redo Log），這種技術能夠在保證數據一致性的同時提高數據訪問效率。為了提高故障恢復（Crash Recovery）的速度，數據庫僅需要回滾某個時間點之后的未寫入磁盤的事物，這個時間點就是檢查點。

（2）快照（Snapshot）：數據的一個拷貝，有兩種實現方式，分別為寫時拷貝（COW，Copy On Write）和寫重定向（ROW，Redirect On Write），其中COW用于讀密集型系統，ROW用于寫密集型系統。

（3）消息隊列pull模式：在分布式消息系統中消費者（Consumer）主動連接緩存代理（Broker）獲取消息的一種消息消費模式。相應地，在push模式中系統將消息主動推送給消費者。

流式數據處理引擎用計算圖的形式編譯數據處理應用程序，其中計算圖用有向無環圖（DAG，Directed Acyclic Execution Graph）的形式描述。它有兩種表示形式，即邏輯形式和物理部署形式。邏輯形式的計算圖由一系列計算節點的單實例組成，而物理部署形式則由計算節點的多個并行實例組成，其中并行實例的含義是在分布式環境中同一計算節點有多個功能相同的物理部署實例，如圖1-34所示，邏輯形式中的map（）節點會有兩個部署實例map（）[1]和map（）[2]。

流式數據處理計算圖中的節點可分為三類：Source（負責數據輸入）、Sink（負責結果輸出）和算子（圖1-34中的map、keyBy和window），它們之間由數據傳輸通道連接。此外，計算節點的每個部署實例也被稱為任務（task）。

以T表示計算節點的集合，E表示邊（數據傳輸通道）的集合，則計算圖可表示為，其抽象形式如圖1-35所示。

以M表示E中傳輸數據的集合，則對于任意一個計算節點：

（1）具有輸入輸出數據集。

（2）具有狀態。

（3）功能由函數[這里指用戶自定義函數（UDF，User Defined Function）]定義。節點拉取數據，由函數更新狀態至，并生成輸出數據，即

一個很自然的想法是對計算圖在某些時間點上做快照，這樣在故障發生后整個數據處理系統可以恢復到某個快照時間點的狀態，以保證exactly-once語義。

定義快照為

其中，是所有節點狀態的集合，即

是所有傳輸通道狀態的集合，即。為了保證exactly-once語義，快照需要具備以下兩個約束條件。

（1）快照必須在有限時間內完成。

（2）快照必須包含所有信息（包括在通道上傳輸的數據）及這些信息的因果關系，這涉及與的關系。為了實現這個看似很自然的想法，先驅者展開了開創性的研究。

（1）同步快照（Global Synchronous Snapshot）。同步快照分為三個步驟：第一步是暫停整個數據處理引擎；第二步是執行快照操作；第三步是繼續執行。同步快照包括此刻仍在所有傳輸通道中的數據和所有節點的狀態，因此快照的容量較大。這種同步機制會嚴重影響系統的吞吐量，增加引擎運行時的系統開銷。（2）異步快照（Asynchronous Snapshot）。為了規避快照同步造成系統吞吐量降低的缺陷，研究人員提出異步機制，即在引擎執行計算任務的同時執行快照操作，且不需要所有節點和傳輸通道同時執行快照操作。這種機制并沒有解決快照容量大的問題，也沒有提升故障恢復效率。

（3）異步屏障快照。這是一種輕量級異步快照，不僅適用于DAG，而且適用于有環圖，本節以DAG為例分析快照算法和故障恢復機制。

ABS的前置條件如下。

（1）傳輸通道提供阻塞（Block）和非阻塞（Unblock）操作，數據以先進先出（FIFO，First Input First Output）的方式傳輸。當傳輸通道處于阻塞狀態時，所有數據將被緩存。

（2）計算節點可以阻塞或和與之連接的通道解除阻塞，并能在通道上傳輸控制消息，還可以在其輸出通道上廣播消息（Broadcasting Messages）。

（3）控制消息流不參與任何節點函數的計算?？煺账惴ǖ牟襟E如下。

（1）引擎定期向 Source 節點插入檢查點屏障（Barrier）。在收到作為控制消息的檢查點屏障后，Source 節點對自己的狀態做快照，并在其輸出通道上廣播此檢查點屏障消息。此外，不同的檢查點屏障可以通過id區分。

（2）當其從任意一個輸入通道收到檢查點屏障消息時，算子或Sink節點阻塞此輸入通道，直至本節點從所有輸入通道收到檢查點屏障。

（3）在其從所有輸入通道收到檢查點屏障后，算子或Sink節點對自己的狀態做快照，然后對其所有輸入通道解除阻塞。

于是，對于同一檢查點屏障，ABS會產生在下面條件時的快照，

ABS算法中條件式（1.1）表明快照僅包括節點的狀態，不包括仍在傳輸通道中的數據，因為這會降低快照的容量。同時，ABS算法很好地兼顧了低延遲和高吞吐。Flink采用ABS算法實現一致性，圖1-36描述了Flink檢查點屏障的流轉過程。

ABS是全局的，這樣可以通過快照id計算出同一時刻計算圖的狀態，引擎可恢復到這些時間點重啟計算任務，進而保證 exactly-once 語義。那么，如何進行故障恢復呢？故障恢復分為以下兩種情況。

（1）計算圖的拓撲結構不變的情況。引擎從持久化后端中讀入上一個可用快照，以此重新初始化所有物理計算任務，即恢復計算圖；然后計算任務繼續運行，就如同沒有發生故障一樣。Flink窗口的狀態不僅包括事件時間推進情況，還包括處理時間推進情況，對于故障恢復這一點異常重要，也從另一個側面說明引入時間與窗口機制加大了架構流式數據處理引擎的難度。

（2）計算圖的拓撲結構發生變化的情況，如圖1-37所示。

這需要引擎的任務管理器（Job Managers/Task Managers）根據快照重新編排計算任務，這也是彈性（Resilient）計算的要求之一。

1.5.4 保存點

檢查點屏障由引擎負責實現，不需要數據處理應用程序編程。保存點（Savepoint）則由應用程序借助檢查點底層機制實現一致性的應用層機制，廣泛應用于數據處理程序平滑升級中。檢查點的目標是輕量，而保存點的目標是實現應用層的一致性功能，例如，Flink 可配置 RocksDB 作為存儲后端以實現增量式狀態，而保存點則不需要這種優化配置。

保存點可用于應用程序平滑升級、引擎升級、A/B 測試等場景。在平滑升級任務中，我們首先保存舊版本程序（如V1.0）在升級前（如時間點t₁）的運行狀態，然后用保存的狀態初始化新版本程序（如 V2.0），這種版本狀態管理是由保存點實現的。此外，為了保證平滑升級和升級失敗回退，在新版本正常提供服務之前，舊版本仍需繼續運行，保存點實現版本狀態管理的過程，如圖1-38所示。