背景

流數據處理，同時提供與批處理對應物相比的諸如新鮮度和更順暢的資源消耗的益處，歷史上與不可靠且具有近似結果的缺點相關。然而，這些缺點不是流媒體數據處理本身的固有特徵，而是如何實現它們先前已經實現的文物。如Google Cloud DataFlow和Apache Beam近年來所示，流處理可以與批處理一樣穩健。在本博客文章中，我們將討論如何使流媒體數據處理強大。這裡有很多想法都是基於磨機，據說谷歌雲數據流建立。

問題分析

人們對流媒體數據處理的魯棒性的主要擔憂似乎圍繞分布式系統中的公共部分失敗情景居中，例如機器故障和網絡中斷。對於批處理，我們可以在部分故障時始終重新運行整個操作，最終重試後批處理將成功。但在流化處理中，重試看起來像什麼？它會丟棄數據或創建重複數據嗎？它會傷害流化處理的可擴展性和性能嗎？

要回答這些問題，我們可以放大單個計算步驟。流數據處理可以被認為是由許多步驟組成的管道。如果我們可以保證單個計算步驟的穩健性，我們可以概括為整個管道。請注意，我們還需要一個跟蹤所有計算並監視整個流水線的邏輯上集中控制服務。但我們不會有時間進入細節。

單個計算步驟

讓我們看一下抽象級別的單個計算步驟，以便我們的討論的結論可以應用於流媒體數據處理中的任何類型的工作。禁止，計算從上游輸入，計算結果，並在下游發送結果，如圖1所示。

> Figure-1

我們首先假設計算沒有外部副作用，如發送電子郵件，遞增外部計數器等。否則我們無法保證流媒體處理本身的魯棒性，因為我們必須能夠在失敗時重新運行計算。如果存在這樣的外部副作用，則通常是一個糟糕的設計標誌。有時，這是必要的，在這種情況下，用戶需要在流處理外部進行副作用。現在我們已經確定了計算可以根據我們想要的多次重試，我們只需要重新運行它，直到我們知道它成功。

接下來，我們需要確保我們不會在舊數據上重新運行它，因為它將產生重複的結果。我們為每個數據點分配唯一ID。計算存儲處理後的數據點ID。每當計算執行時，它將首先檢查其存儲，以確保它以前沒有看到數據點。如果計算已經看到了數據點，則會立即確認上游，以發信號點已處理數據點。顯然，我們必須清理舊數據點ID以保存存儲空間。我們稍後會得到那麼一點。作為延遲優化，我們可以依賴於綻放過濾器來檢查以前的計算是否已看到此數據點。如果數據點ID不在盛開過濾器中，我們知道它是一個新的數據點。Bloom過濾器會大大減少對存儲的查詢。請注意，計算需要根據重新啟動時基於其存儲重新填充綻放過濾器。

最後，我們需要確保我們可靠地在下游交付結果。一旦計算完成處理數據點，它將在其存儲中存儲結果。請注意，計算檢查數據點ID和原子統一地將數據點ID和結果一致至關重要。之後，它在上游確認並將結果下游發送。計算重試，按順序發送結果，直到下游確認。

有關一個計算的整體生命周期和潛在的故障情景，請參閱圖2。

> Figure-2. If the computation fails before (1), it starts overs. The write to storage is atomic so the next potential failure point is (2). After that, the checkpoint is persisted, the computation can restart by reading the checkpoint into memory, and proceed from (2).

有一點值得注意的是，在分布式系統中，沒有可靠地檢測失敗的東西。因此，即使將計算視為失敗，也可能發生這種情況，它實際上仍然是活動的，以可能與新啟動的實例衝突的方式寫入存儲。處理此操作的一種方法是使用寫入租賃來調解對存儲的訪問。控制服務為每個計算分配租約。如果控件服務認為計算已死亡，則它使租賃使租賃失效並為新實例生成新的。由於無效租約，殭屍計算的殘餘寫入將被擊落。

作為吞吐量優化，我們可以撥入數據點信息的處理和存儲。無需單獨做。至於底層存儲系統，Bigtable會符合良好。在Bigtable中寫入單行是原子的。我們可以將單個計算的所有內容作為單個文檔寫入。Bigtable並不能提供強烈的一致性。但無論如何，我們都不需要。我們只需要讀寫的寫一致性，因為計算不需要從其他人的存儲內容中讀取。簽出此博客文章[鏈接]如果您想了解有關分布式系統的一致性的更多信息。

數據清理

我們之前提到過，我們不想無限期地存儲檢查點（數據點ID和結果）。所以我們需要一種機制來清理它們。結果清理很簡單。一旦計算從下游接收到ACK，它就可以丟棄相應的結果。數據點ID清理有點棘手。

要管理數據點ID的生命周期，我們需要在流流程管道中建立時間戳的概念。每個數據點都與時間戳固有相關。它可能是來自外部來源的時間。例如，當流流管線正在將日誌文件作為輸入進行拖尾時，可以使用日誌條目的時間戳。如果我們在流流程管線中生成數據點，它也可能是從內部的時間。例如，當計算產生結果數據點時，它可以將當前壁時間分配為時間戳。

只是為了把炸彈伸出右：無論我們願意等待多久，都可能總是有延遲數據點。想想從手機收集日誌的用例，用戶可以在發送所有本地日誌之前打開飛機模式。我們何時無法預測，最後一架飛機將降落。

因此，在流媒體管道的攝取側，我們必須使用啟發式來確定我們是否已經看到所有數據到達某個時間戳。我們稱之為低水印。例如，我們說我們將等待3天Max，在這種情況下，當前時間減去3天是管道攝入的低水印。對於每個計算步驟，低水印是前一步的低水印的最小值和當前步驟中未處理數據點的最舊的時間戳。有關插圖，請參見圖-3。