Apache Flink 如何管理Kafka消費者offsets

問題導讀

1.Flink與kafka一起如何做Checkpointing ？

2.發生故障，Flink如何恢複的？

3.Kafka consumer offsets存儲在什麼位置？

下面一些詞簡單解釋：

1.檢查點對應Checkpointing

2.主題對應Topic

3.Job對應工作

######################

在我們這篇文章中，我們将逐漸說明Apache Flink如何與Apache Kafka協同工作，以確定Kafka主題（Topic）的記錄exactly-once 保證進行處理。

檢查點（Checkpointing ）是Apache Flink的内部機制，可以從故障中恢複。檢查點是Flink應用程式狀态的一緻副本，包括輸入的讀取位置。如果發生故障，Flink将通過從檢查點加載應用程式狀态并從恢複的讀取位置繼續恢複應用程式，就像沒有發生任何事情一樣。可以将檢查點視為儲存計算機遊戲的目前狀态。如果你在遊戲中儲存了自己的位置後發生了什麼事情，你可以随時回過頭再試一次。

檢查點（Checkpoints ）使Apache Flink具有容錯能力，并確定在發生故障時保留流應用程式的語義。應用程式可以定期觸發檢查點。

Apache Flink中的Kafka消費者将Flink的檢查點機制與有狀态運算符內建在一起，其狀态是所有Kafka分區中的讀取偏移量。觸發檢查點時，每個分區的偏移量都存儲在檢查點中。 Flink的檢查點機制確定所有operator 任務的存儲狀态是一緻的，即它們基于相同的輸入資料。當所有operator 任務成功存儲其狀态時，檢查點完成。是以，當從潛在的系統故障重新啟動時，系統提供一次性狀态更新保證。

下面我們将介紹Apache Flink如何在逐漸指南中檢查Kafka消費者offsets。在我們的示例中，資料存儲在Flink的Job Master中。值得注意的是，在POC或production 用例下，資料通常存儲在外部檔案存儲器（如HDFS或S3）中。

第一步：

下面的示例從Kafka主題中讀取兩個分區，每個分區包含“A”，“B”，“C”，“D”，“E”作為消息。 我們将兩個分區的偏移量設定為零。

第二步：

在第二步中，Kafka消費者開始從分區0讀取消息。消息“A”在 “in-flight”處理，第一個消費者的偏移量變為1。

第三步：

在第三步中，消息“A”到達Flink Map Task。 兩個消費者都讀取他們的下一個記錄（partition 0的消息“B”和partition 1的消息“A”）。 兩個分區的偏移量分别更新為2和1。 與此同時，Flink的Job Master決定在源頭觸發檢查點。

第四步：

在接下來的步驟中，Kafka consumer 任務已經建立了狀态的快照（“offset = 2,1”），現在存儲在Apache Flink的Job Master中。 源分别在來自分區0和1的消息“B”和“A”之後發出檢查點barriers 。 檢查點barriers （障礙）用于對齊所有operator 任務的檢查點，并保證整個檢查點的一緻性。 消息“A”到達Flink Map Task，而top consumer 繼續讀取其下一條記錄（消息“C”）。

第五步：

此步驟顯示Flink Map Task從兩個源和檢查點接收Checkpoints  barriers 。 與此同時，消費者（consumers ）繼續從Kafka分區閱讀更多events 。

第六步：

此步驟顯示Flink Map Task在檢查其狀态後與Flink Job Master進行通信。 當Job 的所有任務确認其狀态為檢查點時， Job Master 完成檢查點。 從現在開始，檢查點可用于從故障中恢複。 值得一提的是，Apache Flink不依賴于Kafka偏移來恢複潛在的系統故障。

在發生故障時恢複

如果發生故障（例如，worker故障），則重新啟動所有operator任務，并将其狀态重置為上次完成的檢查點。

Kafka源分别從偏移量2和1開始，因為這是完成的檢查點的偏移量。 當作業重新啟動時，我們可以期待正常的系統操作，就好像之前沒有發生故障一樣。