1狀態(tài)一致性級別
1.1 AT-MOST-ONCE (最多一次)：
- 這本質(zhì)上是一『盡力而為』的方法。保證數(shù)據(jù)或事件最多由應用程序中的所有算子處理一次。 這意味著如果數(shù)據(jù)在被流應用程序完全處理之前發(fā)生丟失，則不會進行其他重試或者重新發(fā)送。下圖中的例子說明了這種情況。
1.2 AT-LEAST-ONCE (至少一次)：
- 應用程序中的所有算子都保證數(shù)據(jù)或事件至少被處理一次。這通常意味著如果事件在流應用程序完全處理之前丟失，則將從源頭重放或重新傳輸事件。然而，由于事件是可以被重傳的，因此一個事件有時會被處理多次，這就是所謂的至少一次。
下圖的例子描述了這種情況：第一個算子最初未能成功處理事件，然后在重試時成功，接著在第二次重試時也成功了，其實是沒有必要的。
1.3 EXACTLY-ONCE (精確一次)：
即使是在各種故障的情況下，流應用程序中的所有算子都保證事件只會被『精確一次』的處理。（也有文章將 Exactly-once 翻譯為：完全一次，恰好一次）
通常使用兩種流行的機制來實現(xiàn)『精確一次』處理語義。
- 分布式快照 / 狀態(tài)檢查點
- 至少一次事件傳遞和對重復數(shù)據(jù)去重
實現(xiàn)『精確一次』的分布式快照/狀態(tài)檢查點方法受到 Chandy-Lamport 分布式快照算法的啟發(fā)[1]。通過這種機制，流應用程序中每個算子的所有狀態(tài)都會定期做 checkpoint。如果是在系統(tǒng)中的任何地方發(fā)生失敗，每個算子的所有狀態(tài)都回滾到最新的全局一致 checkpoint 點。在回滾期間，將暫停所有處理。源也會重置為與最近 checkpoint 相對應的正確偏移量。整個流應用程序基本上是回到最近一次的一致狀態(tài)，然后程序可以從該狀態(tài)重新啟動。下圖描述了這種 checkpoint 機制的基礎(chǔ)知識。
在上圖中，流應用程序在 T1 時間處正常工作，并且做了checkpoint。然而，在時間 T2，算子未能處理輸入的數(shù)據(jù)。此時，S=4 的狀態(tài)值已保存到持久存儲器中，而狀態(tài)值 S=12 保存在算子的內(nèi)存中。為了修復這種差異，在時間 T3，處理程序?qū)顟B(tài)回滾到 S=4 并“重放”流中的每個連續(xù)狀態(tài)直到最近，并處理每個數(shù)據(jù)。最終結(jié)果是有些數(shù)據(jù)已被處理了多次，但這沒關(guān)系，因為無論執(zhí)行了多少次回滾，結(jié)果狀態(tài)都是相同的。
另一種實現(xiàn)『精確一次』的方法是：在每個算子上實現(xiàn)至少一次事件傳遞和對重復數(shù)據(jù)去重來。使用此方法的流處理引擎將重放失敗事件，以便在事件進入算子中的用戶定義邏輯之前，進一步嘗試處理并移除每個算子的重復事件。此機制要求為每個算子維護一個事務日志，以跟蹤它已處理的事件。利用這種機制的引擎有 Google 的 MillWheel[2] 和 Apache Kafka Streams。下圖說明了這種機制的要點。
這里所說的的狀態(tài)一致性是flink系統(tǒng)內(nèi)部的狀態(tài)，如果要保證端到端，從接受到發(fā)出數(shù)據(jù)都保障一致性，還需要其他系統(tǒng)能力支持，這部分后面說。
絕大多數(shù)情況我們會希望exactly-once，但相比at-least-once，exactly-once的性能與速度會相對較慢一點，這是由于checkpoint的機制造成的。
我們主要關(guān)注的是精確一致性，所以在這里我們也只講精確一致性相關(guān)的概念。
我們所說的flink內(nèi)的精確一致性，真的是精確的只發(fā)送一次數(shù)據(jù)么？
現(xiàn)在讓我們重新審視『精確一次』處理語義真正對最終用戶的保證?！壕_一次』這個術(shù)語在描述正好處理一次時會讓人產(chǎn)生誤導。
有些人可能認為『精確一次』描述了事件處理的保證，其中流中的每個事件只被處理一次。實際上，沒有引擎能夠保證正好只處理一次。在面對任意故障時，不可能保證每個算子中的用戶定義邏輯在每個事件中只執(zhí)行一次，因為用戶代碼被部分執(zhí)行的可能性是永遠存在的。
那么，當引擎聲明『精確一次』處理語義時，它們能保證什么呢？如果不能保證用戶邏輯只執(zhí)行一次，那么什么邏輯只執(zhí)行一次？當引擎聲明『精確一次』處理語義時，它們實際上是在說，它們可以保證引擎管理的狀態(tài)更新只提交一次到持久的后端存儲。
上面描述的兩種機制都使用持久的后端存儲作為真實性的來源，可以保存每個算子的狀態(tài)并自動向其提交更新。對于機制 1 (分布式快照 / 狀態(tài)檢查點)，此持久后端狀態(tài)用于保存流應用程序的全局一致狀態(tài)檢查點(每個算子的檢查點狀態(tài))。對于機制 2 (至少一次事件傳遞加上重復數(shù)據(jù)刪除)，持久后端狀態(tài)用于存儲每個算子的狀態(tài)以及每個算子的事務日志，該日志跟蹤它已經(jīng)完全處理的所有事件。
提交狀態(tài)或?qū)ψ鳛檎鎸崄碓吹某志煤蠖藨酶驴梢员幻枋鰹榍『冒l(fā)生一次。然而，如上所述，計算狀態(tài)的更新 / 更改，即處理在事件上執(zhí)行任意用戶定義邏輯的事件，如果發(fā)生故障，則可能不止一次地發(fā)生。換句話說，事件的處理可以發(fā)生多次，但是該處理的效果只在持久后端狀態(tài)存儲中反映一次。因此，我們認為有效地描述這些處理語義最好的術(shù)語是『有效一次』（effectively once）。
1.4 分布式快照與至少一次事件傳遞和重復數(shù)據(jù)刪除的比較
從語義的角度來看，分布式快照和至少一次事件傳遞以及重復數(shù)據(jù)刪除機制都提供了相同的保證。然而，由于兩種機制之間的實現(xiàn)差異，存在顯著的性能差異。
-
機制 1（分布式快照 / 狀態(tài)檢查點）的性能開銷是最小的，因為引擎實際上是往流應用程序中的所有算子一起發(fā)送常規(guī)事件和特殊事件，而狀態(tài)檢查點可以在后臺異步執(zhí)行。但是，對于大型流應用程序，故障可能會更頻繁地發(fā)生，導致引擎需要暫停應用程序并回滾所有算子的狀態(tài)，這反過來又會影響性能。流式應用程序越大，故障發(fā)生的可能性就越大，因此也越頻繁，反過來，流式應用程序的性能受到的影響也就越大。然而，這種機制是非侵入性的，運行時需要的額外資源影響很小。
-
機制 2（至少一次事件傳遞加重復數(shù)據(jù)刪除）可能需要更多資源，尤其是存儲。使用此機制，引擎需要能夠跟蹤每個算子實例已完全處理的每個元組，以執(zhí)行重復數(shù)據(jù)刪除，以及為每個事件執(zhí)行重復數(shù)據(jù)刪除本身。這意味著需要跟蹤大量的數(shù)據(jù)，尤其是在流應用程序很大或者有許多應用程序在運行的情況下。執(zhí)行重復數(shù)據(jù)刪除的每個算子上的每個事件都會產(chǎn)生性能開銷。但是，使用這種機制，流應用程序的性能不太可能受到應用程序大小的影響。對于機制 1，如果任何算子發(fā)生故障，則需要發(fā)生全局暫停和狀態(tài)回滾；對于機制 2，失敗的影響更加局部性。當在算子中發(fā)生故障時，可能尚未完全處理的事件僅從上游源重放/重傳。性能影響與流應用程序中發(fā)生故障的位置是隔離的，并且對流應用程序中其他算子的性能幾乎沒有影響。從性能角度來看，這兩種機制的優(yōu)缺點如下。
分布式快照 / 狀態(tài)檢查點的優(yōu)缺點：
- 優(yōu)點：
- 較小的性能和資源開銷
- 缺點：
- 對性能的影響較大
- 拓撲越大，對性能的潛在影響越大
至少一次事件傳遞以及重復數(shù)據(jù)刪除機制的優(yōu)缺點：
- 優(yōu)點：
- 故障對性能的影響是局部的
- 故障的影響不一定會隨著拓撲的大小而增加
- 缺點：
- 可能需要大量的存儲和基礎(chǔ)設(shè)施來支持
- 每個算子的每個事件的性能開銷
雖然從理論上講，分布式快照和至少一次事件傳遞加重復數(shù)據(jù)刪除機制之間存在差異，但兩者都可以簡化為至少一次處理加冪等性。對于這兩種機制，當發(fā)生故障時(至少實現(xiàn)一次)，事件將被重放/重傳，并且通過狀態(tài)回滾或事件重復數(shù)據(jù)刪除，算子在更新內(nèi)部管理狀態(tài)時本質(zhì)上是冪等的。
2flink內(nèi)部實現(xiàn)狀態(tài)一致性
從我另一篇文章
Flink checkpoint具體操作流程詳解與報錯調(diào)試方法匯總
中有詳細的闡述了，flink內(nèi)checkpoint的執(zhí)行流程，這里就不展開講了，對這部分知識有了解的讀者應該清楚，節(jié)點做checkpoint的快照部分前是同步的，也就是，這個節(jié)點會等待所有上游并發(fā)節(jié)點的 checkpoint barrier全部到來才會發(fā)出做快照的命令，之后才是異步做快照的階段。
在我們程序處理中通常要求能夠滿足Exactly once語義，保證數(shù)據(jù)的準確性，flink 通過checkpoint機制提供了Exactly-Once與At-Least-Once 兩種不同的消費語義實現(xiàn)， 可以將程序處理的所有數(shù)據(jù)都保存在狀態(tài)內(nèi)部，當程序發(fā)生異常失敗重啟可以從最近一次成功checkpoint中恢復狀態(tài)數(shù)據(jù)，通過checkpoint中barrier對齊機制來實現(xiàn)這兩不同的語義，barrier對齊發(fā)生在一個處理節(jié)點需要接收上游不同處理節(jié)點的數(shù)據(jù)，由于不同的上游節(jié)點數(shù)據(jù)處理速度不一致，那么就會導致下游節(jié)點接收到 barrier的時間點也會不一致，這時候就需要使用barrier對齊機制：在同一checkpoint中，先到達的barrier是否需要等待其他處理節(jié)點barrier達到后在發(fā)送后續(xù)數(shù)據(jù)，barrier將數(shù)據(jù)流分為前后兩個checkpoint(chk n,chk n+1)的概念，如果不等待那么就會導致chk n的階段處理了chk n+1階段的數(shù)據(jù)，但是在source端所記錄的消費偏移量又一致，如果chk n成功之后，后續(xù)的任務處理失敗，任務重啟會消費chk n+1階段數(shù)據(jù)，就會到致數(shù)據(jù)重復消息，如果barrier等待就不會出現(xiàn)這樣情況，因此barrier需要對齊那么就是實現(xiàn)Exactly once語義，否則實現(xiàn)的是at least once語義。由于狀態(tài)是屬于flink內(nèi)部存儲，所以flink 僅僅滿足內(nèi)部Exactly once語義。
至此實現(xiàn)了flink系統(tǒng)內(nèi)部的Exactly once語義。
3 端到端的一致性
端到端的數(shù)據(jù)一致性，主要分三部分
- source
- flink內(nèi)部：這部分由flink內(nèi)部實現(xiàn)，這節(jié)就不提了
- sink
3.1 Source
需要外部源支持可重設(shè)數(shù)據(jù)的讀取位置，例如kafka，或增量保存數(shù)據(jù)的數(shù)據(jù)源，自己記錄offset，例如 mysql 記錄消費到了 多少的id。 kafka consumer作為source，可以將偏移量保存下來，如果后續(xù)任務出現(xiàn)了故障，恢復的時候可以由連接器重置偏移量，重新消費數(shù)據(jù)，保證一致性
3.2 Sink
sink端要保證，任務從故障恢復時，數(shù)據(jù)不會重新寫入到持久化存儲中。一半包括兩種情況：冪等寫入，事務寫入
3.2.1 冪等寫入
所謂冪等操作，是說一個操作，可以重復執(zhí)行很多次，但只導致一次結(jié)果更改，也就是說，后面再重復執(zhí)行就不起作用了。這種一般用于，下游的持久化存儲沒有事務支持的情況，例如redis。
這種情況一般的數(shù)據(jù)格式可以支持重復輸出，例如統(tǒng)計關(guān)注的 uid: uid，重復輸出也不會影響數(shù)據(jù)的準確性
3.2.2 事務寫入
事務寫入需要下游系統(tǒng)支持事務，例如kafka，mysql等。利用flink的兩階段提交，來實現(xiàn)數(shù)據(jù)的exactly-once.
3.2.2.1 兩階段提交
Flink通過checkpoint來保存數(shù)據(jù)是否處理完成的狀態(tài)
由JobManager協(xié)調(diào)各個TaskManager進行checkpoint存儲，checkpoint保存在 StateBackend中，默認StateBackend是內(nèi)存級的，也可以改為文件級的進行持久化保存。
執(zhí)行過程實際上是一個兩段式提交，每個算子執(zhí)行完成**，會進行“預提交”，直到執(zhí)行完sink操作，會發(fā)起“確認提交”，如果執(zhí)行失敗，預提交會放棄掉。**(相當于mysql的事務操作)
如果宕機需要通過StateBackend進行恢復，只能恢復所有未確認提交的操作。
在分布式系統(tǒng)中，可以使用兩階段提交來實現(xiàn)事務性從而保證數(shù)據(jù)的一致性，兩階段提交分為：預提交階段與提交階段，通常包含兩個角色：協(xié)調(diào)者與執(zhí)行者，協(xié)調(diào)者用于用于管理所有執(zhí)行者的操作，執(zhí)行者用于執(zhí)行具體的提交操作，具體的操作流程：
- 首先協(xié)調(diào)者會送預提交(pre-commit)命令有的執(zhí)行者
- 執(zhí)行者執(zhí)行預提交操作然后發(fā)送一條反饋(ack)消息給協(xié)調(diào)者
- 待協(xié)調(diào)者收到所有執(zhí)行者的成功反饋，則發(fā)送一條提交信息(commit)給執(zhí)行者
- 執(zhí)行者執(zhí)行提交操作
如果在流程2中部分預提交失敗，那么協(xié)調(diào)者就會收到一條失敗的反饋，則會發(fā)送一條rollback消息給所有執(zhí)行者，執(zhí)行回滾操作，保證數(shù)據(jù)一致性；但是如果在流程4中，出現(xiàn)部分提交成功部分提交失敗，那么就會造成數(shù)據(jù)的不一致，因此后面也提出了3PC或者通過其他補償機制來保證數(shù)據(jù)最終一致性，接下看看flink 是如何做到2PC，保證數(shù)據(jù)的一致性。
3.2.2.2 flink的兩階段提交
以sink kafka為例，flink兩階段提交步驟詳解：
- 第一條數(shù)據(jù)來了之后，開啟一個kafka的事務(transaction) ，正常寫入kafka分區(qū)日志但標記為未提交，這就是”預提交’
- jobmanager 觸發(fā)checkpoint操作，barrier 從source開始向下傳遞，遇到barrier的算子將狀態(tài)存入狀態(tài)后端,并通知jobmanager
- sink 連接器收到barrier,保存當前狀態(tài)，存入checkpoint,通知jobmanager, 并開啟下一階段的事務，用于提交下個檢查點的數(shù)據(jù)
- jobmanager 收到所有任務的通知，發(fā)出確認信息，表示checkpoint 完成
- sink任務收到j(luò)obmanager的確認信息，正式提交這段時間的數(shù)據(jù)
外部kafka關(guān)閉事務，提交的數(shù)據(jù)可以正常消費了。
如果在這期間出現(xiàn)任何的數(shù)據(jù)問題，flink都會回滾數(shù)據(jù)，之前預提交的數(shù)據(jù)不會被正式寫入到kafka中，但如果沒有問題，也只需要提交一個事務，sink kafka的下游就可以正常消費，sink算子不能存數(shù)據(jù)，這樣的話，數(shù)據(jù)即發(fā)到了下游，有沒有被消費到，出問題又可以回滾，但如果是redis，下游數(shù)據(jù)已經(jīng)寫入到存儲中， 就上flink回滾，寫入的數(shù)據(jù)也無法撤回，這就是兩階段提交的重要性。
個人理解，如果用兩階段提交，數(shù)據(jù)的實時性就沒有那么高，因為需要根據(jù)checkpoint的時間間隔來一批一批的寫入數(shù)據(jù)，沒有辦法每條數(shù)據(jù)都及時的處理完并sink出，所以對實時性有要求的，可以自定義實現(xiàn)冪等性sink來保證數(shù)據(jù)的一致性。
說起來就三部 source flink sink
? Source 引擎可以重新消費，比如 Kafka 可以重置 offset 進行重新消費
? Flink 任務配置 exactly-once，保證 Flink 任務 State 的 exactly-once
? Sink 算子支持兩階段或者可重入，保證產(chǎn)出結(jié)果的 exactly-once
前兩部一半都能支持，關(guān)鍵就是最后sink的步驟
目前第三部一般的做法為：文章來源：http://www.zghlxwxcb.cn/news/detail-834033.html
- ? Sink 兩階段：由于兩階段提交是隨著 Checkpoint 進行的，假設(shè) Checkpoint 是 5min 做一次，那么數(shù)據(jù)對下游消費方的可見性延遲至少也是 5min，所以會有數(shù)據(jù)延遲等問題，目前用的比較少。
- ? Sink 支持可重入：舉例：
- ? Sink 為 MySQL：可以按照 key update 數(shù)據(jù)
- ? Sink 為 Druid：聚合類型可以選用 longMax
- ? Sink 為 ClickHouse：查詢時使用 longMax 或者使用 ReplacingMergeTree 表引擎將重復寫入的數(shù)據(jù)去重，這里有小伙伴會擔心 ReplacingMergeTree 會有性能問題，但是博主認為其實性能影響不會很大，因為 failover 導致的數(shù)據(jù)重復其實一般情況下是小概率事件，并且重復的數(shù)據(jù)量也不會很大，也只是一個 Checkpoint 周期內(nèi)的數(shù)據(jù)重復，所以使用 ReplacingMergeTree 是可以接受的）
- ? Sink 為 Redis：按照 key 更新數(shù)據(jù)
其實就是冪等和兩段式提交 選擇一個文章來源地址http://www.zghlxwxcb.cn/news/detail-834033.html
到了這里，關(guān)于flink如何利用checkpoint保證數(shù)據(jù)狀態(tài)一致性的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
