[[379678]]

本文轉載自微信公眾號「五分鐘學大數(shù)據(jù)」，作者園陌 。轉載本文請聯(lián)系五分鐘學大數(shù)據(jù)公眾號。

Flink

在 Flink 中需要端到端精準一次處理的位置有三個：

Flink 端到端精準一次處理

• Source 端：數(shù)據(jù)從上一階段進入到 Flink 時，需要保證消息精準一次消費。
• Flink 內(nèi)部端：這個我們已經(jīng)了解，利用 Checkpoint 機制，把狀態(tài)存盤，發(fā)生故障的時候可以恢復，保證內(nèi)部的狀態(tài)一致性。不了解的小伙伴可以看下我之前的文章：
• Flink可靠性的基石-checkpoint機制詳細解析

Sink 端：將處理完的數(shù)據(jù)發(fā)送到下一階段時，需要保證數(shù)據(jù)能夠準確無誤發(fā)送到下一階段。

在 Flink 1.4 版本之前，精準一次處理只限于 Flink 應用內(nèi)，也就是所有的 Operator 完全由 Flink 狀態(tài)保存并管理的才能實現(xiàn)精確一次處理。但 Flink 處理完數(shù)據(jù)后大多需要將結果發(fā)送到外部系統(tǒng)，比如 Sink 到 Kafka 中，這個過程中 Flink 并不保證精準一次處理。

在 Flink 1.4 版本正式引入了一個里程碑式的功能：兩階段提交 Sink，即 TwoPhaseCommitSinkFunction 函數(shù)。該 SinkFunction 提取并封裝了兩階段提交協(xié)議中的公共邏輯，自此 Flink 搭配特定 Source 和 Sink(如 Kafka 0.11 版)實現(xiàn)精確一次處理語義(英文簡稱：EOS，即 Exactly-Once Semantics)。

端到端精準一次處理語義(EOS)

以下內(nèi)容適用于 Flink 1.4 及之后版本

對于 Source 端：Source 端的精準一次處理比較簡單，畢竟數(shù)據(jù)是落到 Flink 中，所以 Flink 只需要保存消費數(shù)據(jù)的偏移量即可， 如消費 Kafka 中的數(shù)據(jù)，F(xiàn)link 將 Kafka Consumer 作為 Source，可以將偏移量保存下來，如果后續(xù)任務出現(xiàn)了故障，恢復的時候可以由連接器重置偏移量，重新消費數(shù)據(jù)，保證一致性。

對于 Sink 端：Sink 端是最復雜的，因為數(shù)據(jù)是落地到其他系統(tǒng)上的，數(shù)據(jù)一旦離開 Flink 之后，F(xiàn)link 就監(jiān)控不到這些數(shù)據(jù)了，所以精準一次處理語義必須也要應用于 Flink 寫入數(shù)據(jù)的外部系統(tǒng)，故這些外部系統(tǒng)必須提供一種手段允許提交或回滾這些寫入操作，同時還要保證與 Flink Checkpoint 能夠協(xié)調(diào)使用(Kafka 0.11 版本已經(jīng)實現(xiàn)精確一次處理語義)。

我們以 Flink 與 Kafka 組合為例，F(xiàn)link 從 Kafka 中讀數(shù)據(jù)，處理完的數(shù)據(jù)在寫入 Kafka 中。

為什么以Kafka為例，第一個原因是目前大多數(shù)的 Flink 系統(tǒng)讀寫數(shù)據(jù)都是與 Kafka 系統(tǒng)進行的。第二個原因，也是最重要的原因 Kafka 0.11 版本正式發(fā)布了對于事務的支持，這是與Kafka交互的Flink應用要實現(xiàn)端到端精準一次語義的必要條件。

當然，F(xiàn)link 支持這種精準一次處理語義并不只是限于與 Kafka 的結合，可以使用任何 Source/Sink，只要它們提供了必要的協(xié)調(diào)機制。

Flink 與 Kafka 組合

Flink 應用示例

如上圖所示，F(xiàn)link 中包含以下組件：

1. 一個 Source，從 Kafka 中讀取數(shù)據(jù)(即 KafkaConsumer)
2. 一個時間窗口化的聚會操作(Window)
3. 一個 Sink，將結果寫入到 Kafka(即 KafkaProducer)

若要 Sink 支持精準一次處理語義(EOS)，它必須以事務的方式寫數(shù)據(jù)到 Kafka，這樣當提交事務時兩次 Checkpoint 間的所有寫入操作當作為一個事務被提交。這確保了出現(xiàn)故障或崩潰時這些寫入操作能夠被回滾。

當然了，在一個分布式且含有多個并發(fā)執(zhí)行 Sink 的應用中，僅僅執(zhí)行單次提交或回滾是不夠的，因為所有組件都必須對這些提交或回滾達成共識，這樣才能保證得到一個一致性的結果。Flink 使用兩階段提交協(xié)議以及預提交(Pre-commit)階段來解決這個問題。

兩階段提交協(xié)議(2PC)

兩階段提交協(xié)議(Two-Phase Commit，2PC)是很常用的解決分布式事務問題的方式，它可以保證在分布式事務中，要么所有參與進程都提交事務，要么都取消，即實現(xiàn) ACID 中的 A (原子性)。

在數(shù)據(jù)一致性的環(huán)境下，其代表的含義是：要么所有備份數(shù)據(jù)同時更改某個數(shù)值，要么都不改，以此來達到數(shù)據(jù)的強一致性。

兩階段提交協(xié)議中有兩個重要角色，協(xié)調(diào)者(Coordinator)和參與者(Participant)，其中協(xié)調(diào)者只有一個，起到分布式事務的協(xié)調(diào)管理作用，參與者有多個。

顧名思義，兩階段提交將提交過程劃分為連續(xù)的兩個階段：表決階段(Voting)和提交階段(Commit)。

兩階段提交協(xié)議過程如下圖所示：

兩階段提交協(xié)議

第一階段：表決階段

協(xié)調(diào)者向所有參與者發(fā)送一個 VOTE_REQUEST 消息。

當參與者接收到 VOTE_REQUEST 消息，向協(xié)調(diào)者發(fā)送 VOTE_COMMIT 消息作為回應，告訴協(xié)調(diào)者自己已經(jīng)做好準備提交準備，如果參與者沒有準備好或遇到其他故障，就返回一個 VOTE_ABORT 消息，告訴協(xié)調(diào)者目前無法提交事務。

第二階段：提交階段

• 協(xié)調(diào)者收集來自各個參與者的表決消息。如果所有參與者一致認為可以提交事務，那么協(xié)調(diào)者決定事務的最終提交，在此情形下協(xié)調(diào)者向所有參與者發(fā)送一個 GLOBAL_COMMIT 消息，通知參與者進行本地提交;如果所有參與者中有任意一個返回消息是 VOTE_ABORT，協(xié)調(diào)者就會取消事務，向所有參與者廣播一條 GLOBAL_ABORT 消息通知所有的參與者取消事務。
• 每個提交了表決信息的參與者等候協(xié)調(diào)者返回消息，如果參與者接收到一個 GLOBAL_COMMIT 消息，那么參與者提交本地事務，否則如果接收到 GLOBAL_ABORT 消息，則參與者取消本地事務。

兩階段提交協(xié)議在 Flink 中的應用

Flink 的兩階段提交思路：

我們從 Flink 程序啟動到消費 Kafka 數(shù)據(jù)，最后到 Flink 將數(shù)據(jù) Sink 到 Kafka 為止，來分析 Flink 的精準一次處理。

• 當 Checkpoint 啟動時，JobManager 會將檢查點分界線(checkpoint battier)注入數(shù)據(jù)流，checkpoint barrier 會在算子間傳遞下去，如下如所示：

Flink 精準一次處理：Checkpoint 啟動

• Source 端：Flink Kafka Source 負責保存 Kafka 消費 offset，當 Chckpoint 成功時 Flink 負責提交這些寫入，否則就終止取消掉它們，當 Chckpoint 完成位移保存，它會將 checkpoint barrier(檢查點分界線) 傳給下一個 Operator，然后每個算子會對當前的狀態(tài)做個快照，保存到狀態(tài)后端(State Backend)。

對于 Source 任務而言，就會把當前的 offset 作為狀態(tài)保存起來。下次從 Checkpoint 恢復時，Source 任務可以重新提交偏移量，從上次保存的位置開始重新消費數(shù)據(jù)，如下圖所示：

Flink 精準一次處理：checkpoint barrier 及 offset 保存

• Slink 端：從 Source 端開始，每個內(nèi)部的 transform 任務遇到 checkpoint barrier(檢查點分界線)時，都會把狀態(tài)存到 Checkpoint 里。數(shù)據(jù)處理完畢到 Sink 端時，Sink 任務首先把數(shù)據(jù)寫入外部 Kafka，這些數(shù)據(jù)都屬于預提交的事務(還不能被消費)，此時的 Pre-commit 預提交階段下 Data Sink 在保存狀態(tài)到狀態(tài)后端的同時還必須預提交它的外部事務，如下圖所示：

Flink 精準一次處理：預提交到外部系統(tǒng)

• 當所有算子任務的快照完成(所有創(chuàng)建的快照都被視為是 Checkpoint 的一部分)，也就是這次的 Checkpoint 完成時，JobManager 會向所有任務發(fā)通知，確認這次 Checkpoint 完成，此時 Pre-commit 預提交階段才算完成。才正式到兩階段提交協(xié)議的第二個階段：commit 階段。該階段中 JobManager 會為應用中每個 Operator 發(fā)起 Checkpoint 已完成的回調(diào)邏輯。

本例中的 Data Source 和窗口操作無外部狀態(tài)，因此在該階段，這兩個 Opeartor 無需執(zhí)行任何邏輯，但是 Data Sink 是有外部狀態(tài)的，此時我們必須提交外部事務，當 Sink 任務收到確認通知，就會正式提交之前的事務，Kafka 中未確認的數(shù)據(jù)就改為“已確認”，數(shù)據(jù)就真正可以被消費了，如下圖所示：

Flink 精準一次處理：數(shù)據(jù)精準被消費

注：Flink 由 JobManager 協(xié)調(diào)各個 TaskManager 進行 Checkpoint 存儲，Checkpoint 保存在 StateBackend(狀態(tài)后端) 中，默認 StateBackend 是內(nèi)存級的，也可以改為文件級的進行持久化保存。

最后，一張圖總結下 Flink 的 EOS：