Apache Flink 結合 Kafka 構建端到端的 Exactly-Once 處理

時間 2019-11-05

標籤 apache flink 結合 kafka 構建端的 exactly 處理欄目 Apache 简体版

原文原文鏈接

文章目錄： android

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

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

3. 示例 Flink 應用程序啓動預提交階段網絡

4. 在 Flink 中實現兩階段提交 Operatorsession

5. 總結併發

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

有關 TwoPhaseCommitSinkFunction 的使用詳見文檔: TwoPhaseCommitSinkFunction。或者能夠直接閱讀 Kafka 0.11 sink 的文檔: kafka。異步

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

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

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 讀寫 Kafka ，並提供端到端的 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。

exactly-once-two-phase-commit-2

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

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

exactly-once-two-phase-commit-3

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

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

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

exactly-once-two-phase-commit-4

當 checkpoint barrier 在全部 operator 都傳遞了一遍，而且觸發的 checkpoint 回調成功完成時，預提交階段就結束了。全部觸發的狀態快照都被視爲該 checkpoint 的一部分。checkpoint 是整個應用程序狀態的快照，包括預先提交的外部狀態。若是發生故障，咱們能夠回滾到上次成功完成快照的時間點。

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

數據源和 windnow operator 沒有外部狀態，所以在提交階段，這些 operator 沒必要執行任何操做。可是，數據輸出端（Data Sink）擁有外部狀態，此時應該提交外部事務。

exactly-once-two-phase-commit-5

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

一旦全部 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 語義提供了可能性。
Kafka 0.11 producer的事務是在 TwoPhaseCommitSinkFunction 基礎上實現的，和 at-least-once producer 相比只增長了很是低的開銷。