0 大綱

[Apache Flink]2017年12月發(fā)布的1.4.0版本開始，為流計算引入里程碑特性：TwoPhaseCommitSinkFunction。它提取了兩階段提交協(xié)議的通用邏輯，使得通過Flink來構(gòu)建端到端的Exactly-Once程序成為可能。同時支持：

• 數(shù)據(jù)源（source）
• 和輸出端（sink）

包括Apache Kafka 0.11及更高版本。它提供抽象層，用戶只需實現(xiàn)少數(shù)方法就能實現(xiàn)端到端Exactly-Once語義。

新功能及Flink實現(xiàn)邏輯：

• 描述Flink checkpoint機制如何保證Flink程序結(jié)果的Exactly-Once的
• 顯示Flink如何通過兩階段提交協(xié)議與數(shù)據(jù)源和數(shù)據(jù)輸出端交互，以提供端到端的Exactly-Once保證
• 通過一個簡單的示例，了解如何使用TwoPhaseCommitSinkFunction實現(xiàn)Exactly-Once的文件輸出

1 Flink應(yīng)用中的Exactly-Once語義

Exactly-Once，指每個輸入的事件只影響最終結(jié)果一次。即使機器或軟件故障，既沒有重復(fù)數(shù)據(jù)，也不會丟數(shù)據(jù)。

Flink很久就提供Exactly-Once，checkpoint機制是Flink有能力提供Exactly-Once語義的核心。

一次checkpoint是以下內(nèi)容的一致性快照：

• 應(yīng)用程序的當(dāng)前狀態(tài)
• 輸入流的位置

Flink可配置一個固定時間點，定期產(chǎn)生checkpoint，將checkpoint的數(shù)據(jù)寫入持久存儲系統(tǒng)，如S3或HDFS。將checkpoint數(shù)據(jù)寫入持久存儲是異步，即Flink應(yīng)用程序在checkpoint過程中可以繼續(xù)處理數(shù)據(jù)。

如果發(fā)生機器或軟件故障，重新啟動后，F(xiàn)link應(yīng)用程序?qū)淖钚碌腸heckpoint點恢復(fù)處理； Flink會恢復(fù)應(yīng)用程序狀態(tài)，將輸入流回滾到上次checkpoint保存的位置，然后重新開始運行。這意味著Flink可以像從未發(fā)生過故障一樣計算結(jié)果。

Flink 1.4.0前，Exactly-Once語義僅限Flink應(yīng)用程序內(nèi)部，沒有擴展到Flink數(shù)據(jù)處理完后發(fā)送的大多數(shù)外部系統(tǒng)。Flink應(yīng)用程序與各種數(shù)據(jù)輸出端進行交互，開發(fā)人員自己維護組件上下文保證Exactly-Once語義。

為提供端到端的Exactly-Once語義 – 即除了Flink應(yīng)用程序內(nèi)部，F(xiàn)link寫入的外部系統(tǒng)也需要能滿足Exactly-Once語義 – 這些外部系統(tǒng)必須提供提交或回滾的方法，然后通過Flink的checkpoint機制協(xié)調(diào)。

分布式系統(tǒng)中，協(xié)調(diào)提交和回滾的常用方法是2pc協(xié)議。討論Flink的TwoPhaseCommitSinkFunction如何利用2pc提供端到端的Exactly-Once語義。

2 Flink應(yīng)用程序端到端的Exactly-Once語義

Kafka經(jīng)常與Flink使用。Kafka 0.11版本添加事務(wù)支持。這意味著現(xiàn)在通過Flink讀寫Kafaka，并提供端到端的Exactly-Once語義有了必要支持。

Flink對端到端的Exactly-Once語義的支持不僅局限Kafka，可將它與任何一個提供必要的協(xié)調(diào)機制的源/輸出端一起使用。如Pravega，來自DELL/EMC的開源流媒體存儲系統(tǒng)，通過Flink的TwoPhaseCommitSinkFunction也能支持端到端的Exactly-Once語義。

示例程序有：

• 從Kafka讀取的數(shù)據(jù)源（Flink內(nèi)置的KafkaConsumer）
• 窗口聚合
• 將數(shù)據(jù)寫回Kafka的數(shù)據(jù)輸出端（Flink內(nèi)置的KafkaProducer）

要使數(shù)據(jù)輸出端提供Exactly-Once保證，須將所有數(shù)據(jù)通過一個事務(wù)提交給Kafka。提交捆綁了兩個checkpoint之間的所有要寫數(shù)據(jù)。這確保在故障時，能回滾寫入的數(shù)據(jù)。但分布式系統(tǒng)中，通常有多個并發(fā)運行的寫入任務(wù)，所有組件須在提交或回滾時“一致”才能確保一致結(jié)果。Flink使用2PC及預(yù)提交階段解決這問題。

pre-commit

checkpoint開始時，即2PC的“預(yù)提交”階段。當(dāng)checkpoint開始時，F(xiàn)link的JobManager會將checkpoint barrier（將數(shù)據(jù)流中的記錄分為進入當(dāng)前checkpoint與進入下一個checkpoint）注入數(shù)據(jù)流。

brarrier在operator之間傳遞。對每個operator，它觸發(fā)operator的狀態(tài)快照寫入state backend。

數(shù)據(jù)源保存了消費Kafka的偏移量(offset)，之后將checkpoint barrier傳遞給下一operator。

這種方式僅適用于operator具有『內(nèi)部』狀態(tài)。

內(nèi)部狀態(tài)

指Flink state backend保存和管理的。如第二個operator中window聚合算出來的sum值。當(dāng)一個進程有它的內(nèi)部狀態(tài)時，除了在checkpoint前需將數(shù)據(jù)變更寫入state backend，無需在pre-commit階段執(zhí)行其他操作。

Flink負(fù)責(zé)在checkpoint成功時正確提交這些寫入或故障時中止這些寫入。

3 Flink應(yīng)用啟動pre-commit階段

當(dāng)進程具有『外部』狀態(tài)，需額外處理。外部狀態(tài)通常以寫入外部系統(tǒng)（如Kafka）的形式出現(xiàn)。此時，為提供Exactly-Once保證，外部系統(tǒng)須【支持事務(wù)】，才能和兩階段提交協(xié)議集成。

示例數(shù)據(jù)需寫入Kafka，因此數(shù)據(jù)輸出端（Data Sink）有外部狀態(tài)。此時，在預(yù)提交階段：

• 除了將其狀態(tài)寫入state backend
• 數(shù)據(jù)輸出端還必須預(yù)先提交其外部事務(wù)

當(dāng)checkpoint barrier在所有operator都傳遞了一遍，并且觸發(fā)的checkpoint回調(diào)成功完成時，預(yù)提交階段結(jié)束。所有觸發(fā)的狀態(tài)快照都被視為該checkpoint的一部分。checkpoint是整個應(yīng)用程序狀態(tài)的快照，包括預(yù)先提交的外部狀態(tài)。若故障，可回滾到上次成功完成快照的時間點。

下一步是通知所有operator，checkpoint已經(jīng)成功了。這是2PC的提交階段，JobManager為應(yīng)用程序中的每個operator發(fā)出checkpoint已完成的回調(diào)。

數(shù)據(jù)源和 widnow operator沒有外部狀態(tài)，因此在提交階段，這些operator不必執(zhí)行任何操作。但是，數(shù)據(jù)輸出端（Data Sink）擁有外部狀態(tài)，此時應(yīng)該提交外部事務(wù)。

總結(jié)

• 一旦所有operator完成預(yù)提交，就提交一個commit。
• 如果至少有一個預(yù)提交失敗，則所有其他提交都將中止，我們將回滾到上一個成功完成的checkpoint。
• 在預(yù)提交成功之后，提交的commit需要保證最終成功 – operator和外部系統(tǒng)都需要保障這點。如果commit失?。ɡ纾捎陂g歇性網(wǎng)絡(luò)問題），整個Flink應(yīng)用程序?qū)⑹。瑧?yīng)用程序?qū)⒏鶕?jù)用戶的重啟策略重新啟動，還會嘗試再提交。這個過程至關(guān)重要，因為如果commit最終沒有成功，將會導(dǎo)致數(shù)據(jù)丟失。

因此，我們可以確定所有operator都同意checkpoint的最終結(jié)果：所有operator都同意數(shù)據(jù)已提交，或提交被中止并回滾。

4 在Flink中實現(xiàn)兩階段提交Operator

完整的實現(xiàn)兩階段提交協(xié)議可能有點復(fù)雜，這就是為什么Flink將它的通用邏輯提取到抽象類TwoPhaseCommitSinkFunction中的原因。

接下來基于輸出到文件的簡單示例，說明如何使用TwoPhaseCommitSinkFunction。用戶只需要實現(xiàn)四個函數(shù)，就能為數(shù)據(jù)輸出端實現(xiàn)Exactly-Once語義：

• beginTransaction – 在事務(wù)開始前，我們在目標(biāo)文件系統(tǒng)的臨時目錄中創(chuàng)建一個臨時文件。隨后，我們可以在處理數(shù)據(jù)時將數(shù)據(jù)寫入此文件。
• preCommit – 在預(yù)提交階段，我們刷新文件到存儲，關(guān)閉文件，不再重新寫入。我們還將為屬于下一個checkpoint的任何后續(xù)文件寫入啟動一個新的事務(wù)。
• commit – 在提交階段，我們將預(yù)提交階段的文件原子地移動到真正的目標(biāo)目錄。需要注意的是，這會增加輸出數(shù)據(jù)可見性的延遲。
• abort – 在中止階段，我們刪除臨時文件。

我們知道，如果發(fā)生任何故障，F(xiàn)link會將應(yīng)用程序的狀態(tài)恢復(fù)到最新的一次checkpoint點。一種極端的情況是，預(yù)提交成功了，但在這次commit的通知到達operator之前發(fā)生了故障。在這種情況下，F(xiàn)link會將operator的狀態(tài)恢復(fù)到已經(jīng)預(yù)提交，但尚未真正提交的狀態(tài)。

我們需要在預(yù)提交階段保存足夠多的信息到checkpoint狀態(tài)中，以便在重啟后能正確的中止或提交事務(wù)。在這個例子中，這些信息是臨時文件和目標(biāo)目錄的路徑。

TwoPhaseCommitSinkFunction已經(jīng)把這種情況考慮在內(nèi)了，并且在從checkpoint點恢復(fù)狀態(tài)時，會優(yōu)先發(fā)出一個commit。我們需要以冪等方式實現(xiàn)提交，一般來說，這并不難。在這個示例中，我們可以識別出這樣的情況：臨時文件不在臨時目錄中，但已經(jīng)移動到目標(biāo)目錄了。

在TwoPhaseCommitSinkFunction中，還有一些其他邊界情況也會考慮在內(nèi)，請參考Flink文檔了解更多信息。

FAQ

flink sink在如果過來一個checkpoint barrier，會去存儲state，這個動作會和普通的write并行嗎？還是串行？

在Flink的checkpoint機制中,當(dāng)一個Checkpoint Barrier過來時,sink會觸發(fā)對狀態(tài)的snapshot,這個snapshot動作默認(rèn)是和普通的write操作并行進行的。

具體來說:

• Flink的checkpoint機制是通過在datastream中注入Checkpoint Barrier來實現(xiàn)的。

• 當(dāng)source接收到Checkpoint Barrier時,會將其傳遞給下游的transformation和sink。

• 當(dāng)sink接收到Checkpoint Barrier時,會啟動一個新的線程來執(zhí)行state snapshot(狀態(tài)保存)。

• 這個狀態(tài)snapshot線程會從狀態(tài)后端Snapshot State,并存儲檢查點。

• 而sink的主線程在接收到Checkpoint Barrier時,會繼續(xù)處理正常的write。

• 這樣,狀態(tài)snapshot和正常的write操作就是并行進行的。

但是也可以通過Sink的配置來設(shè)置snapshot和write的執(zhí)行策略,主要有兩種模式:

1. 并行模式(默認(rèn)):snapshot和write同時進行

2. 串行模式:snapshot完成后再進行write

綜上,Flink sink在默認(rèn)的并行checkpoint模式下,狀態(tài)snapshot和普通的write操作是并行執(zhí)行的?？梢酝ㄟ^配置來改變其行為。這樣可以根據(jù)實際需要進行平衡。

總結(jié)

• Flink的checkpoint機制是支持兩階段提交協(xié)議并提供端到端的Exactly-Once語義的基礎(chǔ)。
• 這個方案的優(yōu)點是: Flink不像其他一些系統(tǒng)那樣，通過網(wǎng)絡(luò)傳輸存儲數(shù)據(jù) – 不需要像大多數(shù)批處理程序那樣將計算的每個階段寫入磁盤。
• Flink的TwoPhaseCommitSinkFunction提取了兩階段提交協(xié)議的通用邏輯，基于此將Flink和支持事務(wù)的外部系統(tǒng)結(jié)合，構(gòu)建端到端的Exactly-Once成為可能。
• 從Flink 1.4.0開始，Pravega和Kafka 0.11 producer都提供了Exactly-Once語義；Kafka在0.11版本首次引入了事務(wù)，為在Flink程序中使用Kafka producer提供Exactly-Once語義提供了可能性。
• Kafaka 0.11 producer的事務(wù)是在TwoPhaseCommitSinkFunction基礎(chǔ)上實現(xiàn)的，和at-least-once producer相比只增加了非常低的開銷。

本文由博客一文多發(fā)平臺 OpenWrite 發(fā)布！文章來源地址http://www.zghlxwxcb.cn/news/detail-747178.html

到了這里，關(guān)于Flink實戰(zhàn)(11)-Exactly-Once語義之兩階段提交的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！