Flink的Exactly-once原理是什么

這篇文章主要介紹“Flink的Exactly-once原理是什么”，在日常操作中，相信很多人在Flink的Exactly-once原理是什么問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”Flink的Exactly-once原理是什么”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！

Flink

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

• Source 端：數(shù)據(jù)從上一階段進(jìn)入到 Flink 時，需要保證消息精準(zhǔn)一次消費(fèi)。

• Flink 內(nèi)部端：這個我們已經(jīng)了解，利用 Checkpoint 機(jī)制，把狀態(tài)存盤，發(fā)生故障的時候可以恢復(fù)，保證內(nèi)部的狀態(tài)一致性。不了解的小伙伴可以看下我之前的文章：

  Flink可靠性的基石-checkpoint機(jī)制詳細(xì)解析

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

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

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

端到端精準(zhǔn)一次處理語義（EOS）

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

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

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

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

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

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

Flink 與 Kafka 組合

第一階段：表決階段

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

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

第二階段：提交階段

1. 協(xié)調(diào)者收集來自各個參與者的表決消息。如果所有參與者一致認(rèn)為可以提交事務(wù)，那么協(xié)調(diào)者決定事務(wù)的最終提交，在此情形下協(xié)調(diào)者向所有參與者發(fā)送一個 GLOBAL_COMMIT 消息，通知參與者進(jìn)行本地提交；如果所有參與者中有任意一個返回消息是 VOTE_ABORT，協(xié)調(diào)者就會取消事務(wù)，向所有參與者廣播一條 GLOBAL_ABORT 消息通知所有的參與者取消事務(wù)。

2. 每個提交了表決信息的參與者等候協(xié)調(diào)者返回消息，如果參與者接收到一個 GLOBAL_COMMIT 消息，那么參與者提交本地事務(wù)，否則如果接收到 GLOBAL_ABORT 消息，則參與者取消本地事務(wù)。

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

Flink 的兩階段提交思路：

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

1. 當(dāng) Checkpoint 啟動時，JobManager 會將檢查點(diǎn)分界線（checkpoint battier）注入數(shù)據(jù)流，checkpoint barrier 會在算子間傳遞下去，如下如所示：

Flink 精準(zhǔn)一次處理：checkpoint barrier 及 offset 保存

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

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

到此，關(guān)于“Flink的Exactly-once原理是什么”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識，請繼續(xù)關(guān)注億速云網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬?shí)用的文章！