文章目錄：

Apache Flink 應用程序中的 Exactly-Once 語義

Flink 應用程序端到端的 Exactly-Once 語義

示例 Flink 應用程序啟動預提交階段

在 Flink 中實現兩階段提交 Operator

總結

Apache Flink?自2017年12月發布的1.4.0版本開始，為流計算引入了一個重要的里程碑特性：TwoPhaseCommitSinkFunction（相關的Jira）。它提取了兩階段提交協議的通用邏輯，使得通過Flink來構建端到端的Exactly-Once程序成為可能。同時支持一些數據源（source）和輸出端（sink），包括Apache Kafka 0.11及更高版本。它提供了一個抽象層，用戶只需要實現少數方法就能實現端到端的Exactly-Once語義。

有關TwoPhaseCommitSinkFunction的使用詳見文檔: TwoPhaseCommitSinkFunction。或者可以直接閱讀Kafka 0.11 sink的文檔: kafka。

接下來會詳細分析這個新功能以及Flink的實現邏輯，分為如下幾點。

• 描述Flink checkpoint機制是如何保證Flink程序結果的Exactly-Once的
• 顯示Flink如何通過兩階段提交協議與數據源和數據輸出端交互，以提供端到端的Exactly-Once保證
• 通過一個簡單的示例，了解如何使用TwoPhaseCommitSinkFunction實現Exactly-Once的文件輸出

一、Apache Flink應用程序中的Exactly-Once語義

當我們說『Exactly-Once』時，指的是每個輸入的事件只影響最終結果一次。即使機器或軟件出現故障，既沒有重復數據，也不會丟數據。

Flink很久之前就提供了Exactly-Once語義。在過去幾年中，我們對Flink的checkpoint機制有過深入的描述，這是Flink有能力提供Exactly-Once語義的核心。Flink文檔還提供了該功能的全面概述。

在繼續之前，先看下對checkpoint機制的簡要介紹，這對理解后面的主題至關重要。

• 次checkpoint是以下內容的一致性快照：
• 應用程序的當前狀態
• 輸入流的位置

Flink可以配置一個固定的時間點，定期產生checkpoint，將checkpoint的數據寫入持久存儲系統，例如S3或HDFS。將checkpoint數據寫入持久存儲是異步發生的，這意味著Flink應用程序在checkpoint過程中可以繼續處理數據。

如果發生機器或軟件故障，重新啟動后，Flink應用程序將從最新的checkpoint點恢復處理； Flink會恢復應用程序狀態，將輸入流回滾到上次checkpoint保存的位置，然后重新開始運行。這意味著Flink可以像從未發生過故障一樣計算結果。

在Flink 1.4.0之前，Exactly-Once語義僅限于Flink應用程序內部，并沒有擴展到Flink數據處理完后發送的大多數外部系統。Flink應用程序與各種數據輸出端進行交互，開發人員需要有能力自己維護組件的上下文來保證Exactly-Once語義。

為了提供端到端的Exactly-Once語義 – 也就是說，除了Flink應用程序內部，Flink寫入的外部系統也需要能滿足Exactly-Once語義 – 這些外部系統必須提供提交或回滾的方法，然后通過Flink的checkpoint機制來協調。

分布式系統中，協調提交和回滾的常用方法是兩階段提交協議。在下一節中，我們將討論Flink的TwoPhaseCommitSinkFunction是如何利用兩階段提交協議來提供端到端的Exactly-Once語義。

二、Flink應用程序端到端的Exactly-Once語義

我們將介紹兩階段提交協議，以及它如何在一個讀寫Kafka的Flink程序中實現端到端的Exactly-Once語義。Kafka是一個流行的消息中間件，經常與Flink一起使用。Kafka在最近的0.11版本中添加了對事務的支持。這意味著現在通過Flink讀寫Kafaka，并提供端到端的Exactly-Once語義有了必要的支持。

Flink對端到端的Exactly-Once語義的支持不僅局限于Kafka，您可以將它與任何一個提供了必要的協調機制的源/輸出端一起使用。例如Pravega，來自DELL/EMC的開源流媒體存儲系統，通過Flink的TwoPhaseCommitSinkFunction也能支持端到端的Exactly-Once語義。

在今天討論的這個示例程序中，我們有：

• 從Kafka讀取的數據源（Flink內置的KafkaConsumer）
• 窗口聚合
• 將數據寫回Kafka的數據輸出端（Flink內置的KafkaProducer）

要使數據輸出端提供Exactly-Once保證，它必須將所有數據通過一個事務提交給Kafka。提交捆綁了兩個checkpoint之間的所有要寫入的數據。這可確保在發生故障時能回滾寫入的數據。但是在分布式系統中，通常會有多個并發運行的寫入任務的，簡單的提交或回滾是不夠的，因為所有組件必須在提交或回滾時“一致”才能確保一致的結果。Flink使用兩階段提交協議及預提交階段來解決這個問題。

在checkpoint開始的時候，即兩階段提交協議的“預提交”階段。當checkpoint開始時，Flink的JobManager會將checkpoint barrier（將數據流中的記錄分為進入當前checkpoint與進入下一個checkpoint）注入數據流。

brarrier在operator之間傳遞。對于每一個operator，它觸發operator的狀態快照寫入到state backend。

數據源保存了消費Kafka的偏移量(offset)，之后將checkpoint barrier傳遞給下一個operator。

這種方式僅適用于operator具有『內部』狀態。所謂內部狀態，是指Flink state backend保存和管理的 -例如，第二個operator中window聚合算出來的sum值。當一個進程有它的內部狀態的時候，除了在checkpoint之前需要將數據變更寫入到state backend，不需要在預提交階段執行任何其他操作。Flink負責在checkpoint成功的情況下正確提交這些寫入，或者在出現故障時中止這些寫入。

三、示例Flink應用程序啟動預提交階段

但是，當進程具有『外部』狀態時，需要作些額外的處理。外部狀態通常以寫入外部系統（如Kafka）的形式出現。在這種情況下，為了提供Exactly-Once保證，外部系統必須支持事務，這樣才能和兩階段提交協議集成。

在本文示例中的數據需要寫入Kafka，因此數據輸出端（Data Sink）有外部狀態。在這種情況下，在預提交階段，除了將其狀態寫入state backend之外，數據輸出端還必須預先提交其外部事務。

當checkpoint barrier在所有operator都傳遞了一遍，并且觸發的checkpoint回調成功完成時，預提交階段就結束了。所有觸發的狀態快照都被視為該checkpoint的一部分。checkpoint是整個應用程序狀態的快照，包括預先提交的外部狀態。如果發生故障，我們可以回滾到上次成功完成快照的時間點。

下一步是通知所有operator，checkpoint已經成功了。這是兩階段提交協議的提交階段，JobManager為應用程序中的每個operator發出checkpoint已完成的回調。

數據源和 widnow operator沒有外部狀態，因此在提交階段，這些operator不必執行任何操作。但是，數據輸出端（Data Sink）擁有外部狀態，此時應該提交外部事務。

我們對上述知識點總結下：

• 旦所有operator完成預提交，就提交一個commit。
• 如果至少有一個預提交失敗，則所有其他提交都將中止，我們將回滾到上一個成功完成的checkpoint。
• 在預提交成功之后，提交的commit需要保證最終成功 – operator和外部系統都需要保障這點。如果commit失敗（例如，由于間歇性網絡問題），整個Flink應用程序將失敗，應用程序將根據用戶的重啟策略重新啟動，還會嘗試再提交。這個過程至關重要，因為如果commit最終沒有成功，將會導致數據丟失。

因此，我們可以確定所有operator都同意checkpoint的最終結果：所有operator都同意數據已提交，或提交被中止并回滾。

四、在Flink中實現兩階段提交Operator

完整的實現兩階段提交協議可能有點復雜，這就是為什么Flink將它的通用邏輯提取到抽象類TwoPhaseCommitSinkFunction中的原因。

接下來基于輸出到文件的簡單示例，說明如何使用TwoPhaseCommitSinkFunction。用戶只需要實現四個函數，就能為數據輸出端實現Exactly-Once語義：

• beginTransaction – 在事務開始前，我們在目標文件系統的臨時目錄中創建一個臨時文件。隨后，我們可以在處理數據時將數據寫入此文件。
• preCommit – 在預提交階段，我們刷新文件到存儲，關閉文件，不再重新寫入。我們還將為屬于下一個checkpoint的任何后續文件寫入啟動一個新的事務。
• commit – 在提交階段，我們將預提交階段的文件原子地移動到真正的目標目錄。需要注意的是，這會增加輸出數據可見性的延遲。
• abort – 在中止階段，我們刪除臨時文件。

我們知道，如果發生任何故障，Flink會將應用程序的狀態恢復到最新的一次checkpoint點。一種極端的情況是，預提交成功了，但在這次commit的通知到達operator之前發生了故障。在這種情況下，Flink會將operator的狀態恢復到已經預提交，但尚未真正提交的狀態。

我們需要在預提交階段保存足夠多的信息到checkpoint狀態中，以便在重啟后能正確的中止或提交事務。在這個例子中，這些信息是臨時文件和目標目錄的路徑。

TwoPhaseCommitSinkFunction已經把這種情況考慮在內了，并且在從checkpoint點恢復狀態時，會優先發出一個commit。我們需要以冪等方式實現提交，一般來說，這并不難。在這個示例中，我們可以識別出這樣的情況：臨時文件不在臨時目錄中，但已經移動到目標目錄了。

在TwoPhaseCommitSinkFunction中，還有一些其他邊界情況也會考慮在內，請參考Flink文檔了解更多信息。

總結

總結下本文涉及的一些要點：

• Flink的checkpoint機制是支持兩階段提交協議并提供端到端的Exactly-Once語義的基礎。
• 這個方案的優點是: Flink不像其他一些系統那樣，通過網絡傳輸存儲數據 – 不需要像大多數批處理程序那樣將計算的每個階段寫入磁盤。
• Flink的TwoPhaseCommitSinkFunction提取了兩階段提交協議的通用邏輯，基于此將Flink和支持事務的外部系統結合，構建端到端的Exactly-Once成為可能。
• 從Flink 1.4.0開始，Pravega和Kafka 0.11 producer都提供了Exactly-Once語義；Kafka在0.11版本首次引入了事務，為在Flink程序中使用Kafka producer提供Exactly-Once語義提供了可能性。
• Kafaka 0.11 producer的事務是在TwoPhaseCommitSinkFunction基礎上實現的，和at-least-once producer相比只增加了非常低的開銷。

這是個令人興奮的功能，期待Flink TwoPhaseCommitSinkFunction在未來支持更多的數據接收端。

原文鏈接
本文為云棲社區原創內容，未經允許不得轉載。

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的Apache Flink 结合 Kafka 构建端到端的 Exactly-Once 处理的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。