談談流計算中的『Exactly Once』特性

作者：寶牛

本文翻譯自 streaml.io 網站上的一篇博文：“Exactly once is NOT exactly the same” ，分析了流計算系統中常說的『Exactly Once』特性，主要觀點是：『精确一次』并不保證是完全一樣。主要内容如下：

1. 背景

• 1.1. 最多一次（At-most-once）
• 1.2. 至少一次（At-least-once）
• 1.3. 精确一次（Exactly-once）

1. 『精确一次』是真正的『精确一次』嗎?
2. 分布式快照與至少一次事件傳遞和重複資料删除的比較
3. 結論
4. 參考

目前市面上使用較多的流計算系統有 Apache Storm，Apache Flink, Heron, Apache Kafka (Kafka Streams) 和 Apache Spark (Spark Streaming)。關于流計算系統有個被廣泛讨論的特性是『exactly-once』語義，很多系統宣稱已經支援了這一特性。但是，到底什麼是『exactly-once』，怎麼樣才算是實作了『exactly-once』，人們存在很多誤解和歧義。接下來我們做下分析。

一、背景

流處理（有時稱為事件處理）可以簡單地描述為是對無界資料或事件的連續處理。流或事件處理應用程式可以或多或少地被描述為有向圖，并且通常被描述為有向無環圖（DAG）。在這樣的圖中，每個邊表示資料或事件流，每個頂點表示運算符，會使用程式中定義的邏輯處理來自相鄰邊的資料或事件。有兩種特殊類型的頂點，通常稱為 sources 和 sinks。sources讀取外部資料/事件到應用程式中，而 sinks 通常會收集應用程式生成的結果。下圖是流式應用程式的示例。

A typical stream processing topology

流處理引擎通常允許使用者指定可靠性模式或處理語義，以訓示它将為整個應用程式中的資料處理提供哪些保證。這些保證是有意義的，因為你始終會遇到由于網絡，機器等可能導緻資料丢失的故障。流處理引擎通常為應用程式提供了三種資料處理語義：最多一次、至少一次和精确一次。

如下是對這些不同處理語義的寬松定義：

最多一次（At-most-once)

這本質上是一『盡力而為』的方法。保證資料或事件最多由應用程式中的所有算子處理一次。 這意味着如果資料在被流應用程式完全處理之前發生丢失，則不會進行其他重試或者重新發送。下圖中的例子說明了這種情況。

At-most-once processing semantics

至少一次（At-least-once）

應用程式中的所有算子都保證資料或事件至少被處理一次。這通常意味着如果事件在流應用程式完全處理之前丢失，則将從源頭重放或重新傳輸事件。然而，由于事件是可以被重傳的，是以一個事件有時會被處理多次，這就是所謂的至少一次。

下圖的例子描述了這種情況：第一個算子最初未能成功處理事件，然後在重試時成功，接着在第二次重試時也成功了，其實是沒有必要的。

At-least-once processing semantics

精确一次（Exactly-once）

即使是在各種故障的情況下，流應用程式中的所有算子都保證事件隻會被『精确一次』的處理。（也有文章将 Exactly-once 翻譯為：完全一次，恰好一次）

通常使用兩種流行的機制來實作『精确一次』處理語義。

• 分布式快照 / 狀态檢查點
• 至少一次事件傳遞和對重複資料去重

實作『精确一次』的分布式快照/狀态檢查點方法受到 Chandy-Lamport 分布式快照算法的啟發[1]。通過這種機制，流應用程式中每個算子的所有狀态都會定期做 checkpoint。如果是在系統中的任何地方發生失敗，每個算子的所有狀态都復原到最新的全局一緻 checkpoint 點。在復原期間，将暫停所有處理。源也會重置為與最近 checkpoint 相對應的正确偏移量。整個流應用程式基本上是回到最近一次的一緻狀态，然後程式可以從該狀态重新啟動。下圖描述了這種 checkpoint 機制的基礎知識。

Distributed snapshot

在上圖中，流應用程式在 T1 時間處正常工作，并且做了checkpoint。然而，在時間 T2，算子未能處理輸入的資料。此時，S=4 的狀态值已儲存到持久存儲器中，而狀态值 S=12 儲存在算子的記憶體中。為了修複這種差異，在時間 T3，處理程式将狀态復原到 S=4 并“重放”流中的每個連續狀态直到最近，并處理每個資料。最終結果是有些資料已被處理了多次，但這沒關系，因為無論執行了多少次復原，結果狀态都是相同的。

另一種實作『精确一次』的方法是：在每個算子上實作至少一次事件傳遞和對重複資料去重來。使用此方法的流處理引擎将重放失敗事件，以便在事件進入算子中的使用者定義邏輯之前，進一步嘗試處理并移除每個算子的重複事件。此機制要求為每個算子維護一個事務日志，以跟蹤它已處理的事件。利用這種機制的引擎有 Google 的 MillWheel[2] 和 Apache Kafka Streams。下圖說明了這種機制的要點。

At-least-once delivery plus deduplication

二、『精确一次』是真正的『精确一次』嗎?

現在讓我們重新審視『精确一次』處理語義真正對最終使用者的保證。『精确一次』這個術語在描述正好處理一次時會讓人産生誤導。

有些人可能認為『精确一次』描述了事件處理的保證，其中流中的每個事件隻被處理一次。實際上，沒有引擎能夠保證正好隻處理一次。在面對任意故障時，不可能保證每個算子中的使用者定義邏輯在每個事件中隻執行一次，因為使用者代碼被部分執行的可能性是永遠存在的。

考慮具有流處理運算符的場景，該運算符執行列印傳入事件的 ID 的映射操作，然後傳回事件不變。下面的僞代碼說明了這個操作：

Map (Event event) {
    Print "Event ID: " + event.getId()
    Return event
}           

每個事件都有一個 GUID (全局惟一ID)。如果使用者邏輯的精确執行一次得到保證，那麼事件 ID 将隻輸出一次。但是，這是無法保證的，因為在使用者定義的邏輯的執行過程中，随時都可能發生故障。引擎無法自行确定執行使用者定義的處理邏輯的時間點。是以，不能保證任意使用者定義的邏輯隻執行一次。這也意味着，在使用者定義的邏輯中實作的外部操作(如寫資料庫)也不能保證隻執行一次。此類操作仍然需要以幂等的方式執行。

那麼，當引擎聲明『精确一次』處理語義時，它們能保證什麼呢？如果不能保證使用者邏輯隻執行一次，那麼什麼邏輯隻執行一次？當引擎聲明『精确一次』處理語義時，它們實際上是在說，它們可以保證引擎管理的狀态更新隻送出一次到持久的後端存儲。

上面描述的兩種機制都使用持久的後端存儲作為真實性的來源，可以儲存每個算子的狀态并自動向其送出更新。對于機制 1 (分布式快照 / 狀态檢查點)，此持久後端狀态用于儲存流應用程式的全局一緻狀态檢查點(每個算子的檢查點狀态)。對于機制 2 (至少一次事件傳遞加上重複資料删除)，持久後端狀态用于存儲每個算子的狀态以及每個算子的事務日志，該日志跟蹤它已經完全處理的所有事件。

送出狀态或對作為真實來源的持久後端應用更新可以被描述為恰好發生一次。然而，如上所述，計算狀态的更新 / 更改，即處理在事件上執行任意使用者定義邏輯的事件，如果發生故障，則可能不止一次地發生。換句話說，事件的處理可以發生多次，但是該處理的效果隻在持久後端狀态存儲中反映一次。是以，我們認為有效地描述這些處理語義最好的術語是『有效一次』（effectively once）。

三、分布式快照與至少一次事件傳遞和重複資料删除的比較

從語義的角度來看，分布式快照和至少一次事件傳遞以及重複資料删除機制都提供了相同的保證。然而，由于兩種機制之間的實作差異，存在顯着的性能差異。

機制 1（分布式快照 / 狀态檢查點）的性能開銷是最小的**，因為引擎實際上是往流應用程式中的所有算子一起發送正常事件和特殊事件，而狀态檢查點可以在背景異步執行。但是，對于大型流應用程式，故障可能會更頻繁地發生，導緻引擎需要暫停應用程式并復原所有算子的狀态，這反過來又會影響性能。流式應用程式越大，故障發生的可能性就越大，是以也越頻繁，反過來，流式應用程式的性能受到的影響也就越大。然而，這種機制是非侵入性的，運作時需要的額外資源影響很小。

機制 2（至少一次事件傳遞加重複資料删除）可能需要更多資源，尤其是存儲**。使用此機制，引擎需要能夠跟蹤每個算子執行個體已完全處理的每個元組，以執行重複資料删除，以及為每個事件執行重複資料删除本身。這意味着需要跟蹤大量的資料，尤其是在流應用程式很大或者有許多應用程式在運作的情況下。執行重複資料删除的每個算子上的每個事件都會産生性能開銷。但是，使用這種機制，流應用程式的性能不太可能受到應用程式大小的影響。對于機制 1，如果任何算子發生故障，則需要發生全局暫停和狀态復原；對于機制 2，失敗的影響更加局部性。當在算子中發生故障時，可能尚未完全處理的事件僅從上遊源重放/重傳。性能影響與流應用程式中發生故障的位置是隔離的，并且對流應用程式中其他算子的性能幾乎沒有影響。從性能角度來看，這兩種機制的優缺點如下。

分布式快照 / 狀态檢查點的優缺點：

• 優點：
• 較小的性能和資源開銷
• 缺點：
• 對性能的影響較大
• 拓撲越大，對性能的潛在影響越大

至少一次事件傳遞以及重複資料删除機制的優缺點：

• 故障對性能的影響是局部的
• 故障的影響不一定會随着拓撲的大小而增加
• 可能需要大量的存儲和基礎設施來支援
• 每個算子的每個事件的性能開銷

雖然從理論上講，分布式快照和至少一次事件傳遞加重複資料删除機制之間存在差異，但兩者都可以簡化為至少一次處理加幂等性。對于這兩種機制，當發生故障時(至少實作一次)，事件将被重放/重傳，并且通過狀态復原或事件重複資料删除，算子在更新内部管理狀态時本質上是幂等的。

四、結論

在這篇部落格文章中，我希望能夠讓你相信『精确一次』這個詞是非常具有誤導性的。提供『精确一次』的處理語義實際上意味着流處理引擎管理的算子狀态的不同更新隻反映一次。『精确一次』并不能保證事件的處理，即任意使用者定義邏輯的執行，隻會發生一次。我們更喜歡用『有效一次』（effectively once）這個術語來表示這種保證，因為處理不一定保證隻發生一次，但是對引擎管理的狀态的影響隻反映一次。兩種流行的機制，分布式快照和重複資料删除，被用來實作精确/有效的一次性處理語義。這兩種機制為消息處理和狀态更新提供了相同的語義保證，但是在性能上存在差異。這篇文章并不是要讓你相信任何一種機制都優于另一種，因為它們各有利弊。

五、參考

1. Chandy, K. Mani and Leslie Lamport.Distributed snapshots: Determining global states of distributed systems. ACMTransactions on Computer Systems (TOCS) 3.1 (1985): 63-75.
2. Akidau, Tyler, et al. MillWheel:Fault-tolerant stream processing at internet scale. Proceedings of the VLDBEndowment 6.11 (2013): 1033-1044.