Flink是怎么保證端到端exactly-once語義

Flink是怎么保證端到端exactly-once語義，很多新手對此不是很清楚，為了幫助大家解決這個難題，下面小編將為大家詳細講解，有這方面需求的人可以來學習下，希望你能有所收獲。

2017年12月，apache flink 1.4.0發布。其中有一個里程碑式的功能：兩部提交的sink function（TwoPhaseCommitSinkFunction）。

TwoPhaseCommitSinkFunction 就是把最后寫入存儲的邏輯分為兩部提交，這樣就有可能構建一個從數據源到數據輸出的一個端到端的exactly-once語義的flink應用。當然，TwoPhaseCommitSinkFunction的數據輸出包括apache kafka 0.11以上的版本。flink提供了一個抽象的TwoPhaseCommitSinkFunction類，來讓開發者用更少的代碼來實現端到端的exactly-once語義。

接下來，我們進一步介紹flink的這個特性：

    ?flink的checkpoints在保證exactly-once語義是的作用    ?flink是如何通過兩部提交協議來保證從數據源到數據輸出的exactly-once語義    ?通過一個例子來解釋如果應用TwoPhaseCommitSinkFunction來實現一個exactly-once的sink

exactly-once語義就是保證最后的數據處理的結果和數據攝入時沒有數據的丟失與重復。flink的checkpoint包含了，flink應用現在的狀態與數據輸入流的位置（對于kafka來說就是offset） checkpoint可異步的持久化到像s3或者hdfs這樣子的存儲系統里面。如果flink應用失敗或者升級時，可以拉取checkpoint中的狀態來恢復上次失敗時的數據。在flink1.4.0之前，flink通過checkpoint保證了flink應用內部的exactly-once語義。現在加入了TwoPhaseCommitSinkFunctio可以保證端到端的exactly-once語義。兩次提交來保證語義的方式需要flink所連接的外部系統支持兩部提交，也就是外部系統要支持可以預提交和回滾沒有最終提交的數據這樣子的特性。后面我們會flink是如何與外部系統進行二次提交協議來保證語義的 使用flink來保證端到端的數據不丟失不重復 下面我們來看看flink消費并寫入kafka的例子是如何通過兩部提交來保證exactly-once語義的。kafka從0.11開始支持事物操作，若要使用flink端到端exactly-once語義需要flink的sink的kafka是0.11版本以上的。同時 DELL/EMC的 Pravega 也支持使用flink來保證端到端的exactly-once語義。這個例子包括以下幾個步驟：

    ?從kafka讀取數據    ?一個聚合窗操作    ?向kafka寫入數據

為了保證exactly-once，所有寫入kafka的操作必須是事物的。在兩次checkpiont之間要批量提交數據，這樣在任務失敗后就可以將沒有提交的數據回滾。然而一個簡單的提交和回滾，對于一個分布式的流式數據處理系統來說是遠遠不夠的。

兩部提交協議的第一步是預提交。flink的jobmanager會在數據流中插入一個檢查點的標記（這個標記可以用來區別這次checkpoint的數據和下次checkpoint的數據）。這個標記會在整個dag中傳遞。每個dag中的算子遇到這個標記就會觸發這個算子狀態的快照。

讀取kafka的算子，在遇到檢查點標記時會存儲kafka的offset。之后，會把這個檢查點標記傳到下一個算子。 接下來就到了flink的內存操作算子。這些內部算子就不用考慮兩部提交協議了，因為他們的狀態會隨著flink整體的狀態來更新或者回滾。

到了和外部系統打交道的時候，就需要兩步提交協議來保證數據不丟失不重復了。在預提交這個步驟下，所有向kafka提交的數據都是預提交。

當所有算子的快照完成，也就是這次的checkpoint完成時，flink的jobmanager會向所有算子發通知說這次checkpoint完成，flink負責向kafka寫入數據的算子也會正式提交之前寫操作的數據。在任務運行中的任何階段失敗，都會從上一次的狀態恢復，所有沒有正式提交的數據也會回滾。

5 總結一下flink的兩步提交：

    ?當所有算子都完成他們的快照時，進行正式提交操作    ?當任意子任務在預提交階段失敗時，其他任務立即停止，并回滾到上一次成功快照的狀態。    ?在預提交狀態成功后，外部系統需要完美支持正式提交之前的操作。如果有提交失敗發生，整個

flink應用會進入失敗狀態并重啟，重啟后將會繼續從上次狀態來嘗試進行提交操作。在flink中應用兩步提交算子 在使用兩步提交算子時，我們可以繼承TwoPhaseCommitSinkFunction這個虛擬類。

通過一個簡單的寫文件的例子來解釋一下這個虛擬類。這個兩步提交的類有四個狀態。

    ?1.開始事物（beginTransaction）- 創建一個臨時文件夾，來寫把數據寫入到這個文件夾里面。    ?2.預提交（preCommit）- 將內存中緩存的數據寫入文件并關閉。    ?3.正式提交（commit）- 將之前寫完的臨時文件放入目標目錄下。這代表著最終的數據會有一些延遲。    ?4.丟棄（abort）- 丟棄臨時文件。

若失敗發生在預提交成功后，正式提交前。可以根據狀態來提交預提交的數據，也可刪除預提交的數據。

總結 flink通過狀態和兩次提交協議來保證了端到端的exactly-once語義。在批次處理時，flink不用把每一次的計算都持久話到內存（這句話沒有太理解） flink支持Pravega和 kafka 0.11版本之上的 生產者的exactly-once語義的保證。

看完上述內容是否對您有幫助呢？如果還想對相關知識有進一步的了解或閱讀更多相關文章，請關注億速云行業資訊頻道，感謝您對億速云的支持。