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

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

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

TwoPhaseCommitSinkFunction 就是把最后寫入存儲的邏輯分為兩部提交，這樣就有可能構(gòu)建一個(gè)從數(shù)據(jù)源到數(shù)據(jù)輸出的一個(gè)端到端的exactly-once語義的flink應(yīng)用。當(dāng)然，TwoPhaseCommitSinkFunction的數(shù)據(jù)輸出包括apache kafka 0.11以上的版本。flink提供了一個(gè)抽象的TwoPhaseCommitSinkFunction類，來讓開發(fā)者用更少的代碼來實(shí)現(xiàn)端到端的exactly-once語義。

接下來，我們進(jìn)一步介紹flink的這個(gè)特性：

    ?flink的checkpoints在保證exactly-once語義是的作用    ?flink是如何通過兩部提交協(xié)議來保證從數(shù)據(jù)源到數(shù)據(jù)輸出的exactly-once語義    ?通過一個(gè)例子來解釋如果應(yīng)用TwoPhaseCommitSinkFunction來實(shí)現(xiàn)一個(gè)exactly-once的sink

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

    ?從kafka讀取數(shù)據(jù)    ?一個(gè)聚合窗操作    ?向kafka寫入數(shù)據(jù)

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

兩部提交協(xié)議的第一步是預(yù)提交。flink的jobmanager會在數(shù)據(jù)流中插入一個(gè)檢查點(diǎn)的標(biāo)記（這個(gè)標(biāo)記可以用來區(qū)別這次checkpoint的數(shù)據(jù)和下次checkpoint的數(shù)據(jù)）。這個(gè)標(biāo)記會在整個(gè)dag中傳遞。每個(gè)dag中的算子遇到這個(gè)標(biāo)記就會觸發(fā)這個(gè)算子狀態(tài)的快照。

讀取kafka的算子，在遇到檢查點(diǎn)標(biāo)記時(shí)會存儲kafka的offset。之后，會把這個(gè)檢查點(diǎn)標(biāo)記傳到下一個(gè)算子。 接下來就到了flink的內(nèi)存操作算子。這些內(nèi)部算子就不用考慮兩部提交協(xié)議了，因?yàn)樗麄兊臓顟B(tài)會隨著flink整體的狀態(tài)來更新或者回滾。

到了和外部系統(tǒng)打交道的時(shí)候，就需要兩步提交協(xié)議來保證數(shù)據(jù)不丟失不重復(fù)了。在預(yù)提交這個(gè)步驟下，所有向kafka提交的數(shù)據(jù)都是預(yù)提交。

當(dāng)所有算子的快照完成，也就是這次的checkpoint完成時(shí)，flink的jobmanager會向所有算子發(fā)通知說這次checkpoint完成，flink負(fù)責(zé)向kafka寫入數(shù)據(jù)的算子也會正式提交之前寫操作的數(shù)據(jù)。在任務(wù)運(yùn)行中的任何階段失敗，都會從上一次的狀態(tài)恢復(fù)，所有沒有正式提交的數(shù)據(jù)也會回滾。

5 總結(jié)一下flink的兩步提交：

    ?當(dāng)所有算子都完成他們的快照時(shí)，進(jìn)行正式提交操作    ?當(dāng)任意子任務(wù)在預(yù)提交階段失敗時(shí)，其他任務(wù)立即停止，并回滾到上一次成功快照的狀態(tài)。    ?在預(yù)提交狀態(tài)成功后，外部系統(tǒng)需要完美支持正式提交之前的操作。如果有提交失敗發(fā)生，整個(gè)

flink應(yīng)用會進(jìn)入失敗狀態(tài)并重啟，重啟后將會繼續(xù)從上次狀態(tài)來嘗試進(jìn)行提交操作。在flink中應(yīng)用兩步提交算子 在使用兩步提交算子時(shí)，我們可以繼承TwoPhaseCommitSinkFunction這個(gè)虛擬類。

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

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

若失敗發(fā)生在預(yù)提交成功后，正式提交前?？梢愿鶕?jù)狀態(tài)來提交預(yù)提交的數(shù)據(jù)，也可刪除預(yù)提交的數(shù)據(jù)。

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

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