一垫卤、Checkpoint
獲取分布式數(shù)據(jù)流和算子狀態(tài)的一致性快照是Flink容錯機(jī)制的核心闪檬，這些快照在Flink作業(yè)恢復(fù)時作為一致性檢查點(diǎn)存在嫁审。
1.1 原理
1.1.1 Barriers
Barrier是由流數(shù)據(jù)源（stream source）注入數(shù)據(jù)流中，并作為數(shù)據(jù)流的一部分與數(shù)據(jù)記錄一起往下游流動傻丝。Barriers將流里的記錄分隔為一段一段的記錄集，每一個記錄集都對應(yīng)一個快照诉儒。每個Barrier會攜帶一個快照的ID葡缰，這個快照對應(yīng)Barrier前面的記錄集。如下圖所示忱反。
當(dāng)一個算子從所有輸入流都接收到一個快照（n）的barrier時泛释，它首先會生成該算子的狀態(tài)快照，然后往該算子的所有下游廣播一個barrier温算。這個算子是sink算子時怜校，它會告知檢查點(diǎn)的 coordinator（一般是flink的jobManager）,當(dāng)所有sink算子都告知接收到了一個快照的barrier時，該快照生成結(jié)束注竿。
1.1.2 對齊檢查點(diǎn)（aligned Checkpoint)
當(dāng)一個算子接收到多于一個輸入流時茄茁，就需要進(jìn)行這些流的barrier對齊。當(dāng)一個算子接收到第一個輸入流的快照barrier n時巩割，它不能繼續(xù)處理該流的其他數(shù)據(jù)裙顽，而是需要等待接收到最后一個流的barrier n，才可以生成算子的狀態(tài)快照和發(fā)送掛起的輸出記錄宣谈，然后發(fā)送快照barrier n愈犹。否則，檢查點(diǎn)快照n跟檢查點(diǎn)n+1就會混淆闻丑。
檢查點(diǎn)對齊保證了狀態(tài)的準(zhǔn)確性漩怎，但由于對齊操作是阻塞式的，會使檢查點(diǎn)生成時長不可控嗦嗡，降低吞吐量勋锤，當(dāng)作業(yè)出現(xiàn)反壓時，會加劇反壓酸钦，甚至導(dǎo)致作業(yè)不穩(wěn)定等問題怪得。
1.1.3 非對齊檢查點(diǎn)（Unaligned Checkpoint)
為規(guī)避上述風(fēng)險咱枉，從Flink 1.11開始，檢查點(diǎn)也可以是非對齊的徒恋。具體方法比較類似于Chandy-Lamport算法蚕断，但Flink仍會在數(shù)據(jù)源插入barrier來避免檢查點(diǎn)coordinator負(fù)載過重。
具體處理過程是這樣的：算子在接收到第一個數(shù)據(jù)流的barrier n時就立即開始生成快照入挣，將該barrier發(fā)往下游亿乳，將其他流中后續(xù)到來的該快照的記錄進(jìn)行異步存儲，并創(chuàng)建各自的狀態(tài)快照径筏。
非對齊檢查點(diǎn)可以保證barrier盡快到達(dá)sink, 非常適合算子拓?fù)渲兄辽儆幸粭l緩慢路徑的場景葛假。然而，由于會增大I/O壓力滋恬，如果寫入狀態(tài)后端是處理瓶頸的話聊训，使用非對齊檢查點(diǎn)就不太合適了。
1.2 使用配置
檢查點(diǎn)的配置有許多項恢氯，下面就介紹幾個比較常用的重要配置带斑。
1.2.1 檢查點(diǎn)保留策略
檢查點(diǎn)在默認(rèn)不保留，而且只用于作業(yè)失敗時的恢復(fù)勋拟。程序取消時勋磕，檢查點(diǎn)會被刪除。但可以通過配置周期性的保留檢查點(diǎn)敢靡，由于這些配置挂滓，當(dāng)作業(yè)失敗或者取消時，保留的檢查點(diǎn)不會被自動清除啸胧。這樣作業(yè)失敗時赶站，就有一個可以用于恢復(fù)的檢查點(diǎn)了。
CheckpointConfig config = env.getCheckpointConfig();
config.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
ExternalizedCheckpointCleanup配置了作業(yè)取消時檢查點(diǎn)是保留還是刪除吓揪。
- ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION： 取消作業(yè)時保留檢查點(diǎn)亲怠。請注意，在這種情況下柠辞，您必須在取消后手動清理檢查點(diǎn)狀態(tài)团秽。
- ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION： 取消作業(yè)時刪除檢查點(diǎn)。只有在作業(yè)失敗時叭首，檢查點(diǎn)狀態(tài)才可用习勤。
1.2.2 目錄配置
檢查點(diǎn)保存的信息包括元數(shù)據(jù)文件和數(shù)據(jù)文件，可以通過配置文件配置, 并且是全局性的焙格，對所有作業(yè)適用图毕。
state.checkpoints.dir: hdfs:///checkpoints/
在使用狀態(tài)后端FsStateBackend或者RocksDBStateBackend時，可以指定相關(guān)作業(yè)的檢查點(diǎn)保存目錄
env.setStateBackend(new RocksDBStateBackend("hdfs:///checkpoints-data/"));
1.2.3 模式配置
Checkpoint支持兩種模式：exactly-once和at-least-once眷唉， 默認(rèn)模式為exactly-once予颤。
// set mode to exactly-once (this is the default)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
- Exactly Once：保證每條數(shù)據(jù)對于 Flink 的狀態(tài)結(jié)果只影響一次囤官。
- At Least Once：每條數(shù)據(jù)對于 Flink 狀態(tài)計算至少影響一次。
1.2.4 backend選擇
關(guān)于backend配置及選擇可以參考上一篇Flink詳解之七--狀態(tài)管理
1.3 非對齊檢查點(diǎn)(unaligned checkpoint)
二蛤虐、Savepoint
savepoint是使用檢查點(diǎn)機(jī)制創(chuàng)建的党饮，作業(yè)執(zhí)行狀態(tài)的全局鏡像，可用于flink的停止與恢復(fù)驳庭，升級等刑顺。savepoint有兩部分構(gòu)成：一是在穩(wěn)定存儲（如：HDFS、S3等）中保存了二進(jìn)制文件的目錄饲常，二是元數(shù)據(jù)文件蹲堂。這些文件表示了作業(yè)執(zhí)行狀態(tài)的鏡像，其中元數(shù)據(jù)文件主要保存了以絕對路徑表示的指針贝淤。
savepoint文件默認(rèn)保存目錄配置
# Default savepoint target directory
state.savepoints.dir: hdfs:///flink/savepoints
1.1 使用
# 觸發(fā)savepoint
$ bin/flink savepoint :jobId [:targetDirectory]
#刪除savepoint
$ bin/flink cancel -s [:targetDirectory] :jobId
#從savepoint恢復(fù)
$ bin/flink run -s :savepointPath [:runArgs]
#刪除savepoint
$ bin/flink savepoint -d :savepointPath
1.2 配置算子ID
使用savepoint進(jìn)行恢復(fù)時柒竞，是根據(jù)算子ID來匹配算子狀態(tài)在savepoint中的存儲位置的。官方文檔強(qiáng)烈建議給每個算子手動配置一個算子ID霹娄，這個ID可以通過uid(String)方法配置能犯。
當(dāng)沒有手動配置時，程序會根據(jù)算子在的程序算子拓?fù)渲械奈恢蒙梢粋€ID犬耻。如果程序沒有改變，是可以從savepoint中恢復(fù)的执泰，但如果程序改變了枕磁，同一個算子在程序中的位置也就改變了，相應(yīng)的算子ID也會變化术吝，就無法從之前的savepoint恢復(fù)了计济。
DataStream<String> stream = env.
// Stateful source (e.g. Kafka) with ID
.addSource(new StatefulSource())
.uid("source-id") // ID for the source operator
.shuffle()
// Stateful mapper with ID
.map(new StatefulMapper())
.uid("mapper-id") // ID for the mapper
// Stateless printing sink
.print(); // Auto-generated ID
由于有些算子是有狀態(tài)的，有些算子是無狀態(tài)的排苍，實(shí)際上只要給有狀態(tài)的算子添加算子ID就可以沦寂，但很多人并不太清楚哪些算子是有狀態(tài)的，哪些是無狀態(tài)的淘衙，所以传藏，實(shí)際操作中，盡量給每個算子手動配置算子ID彤守。
三毯侦、Checkpoint與Savepoint的異同
3.1 相同點(diǎn)
- 都用于作業(yè)的恢復(fù)
- 創(chuàng)建時都使用ckeckpoint機(jī)制，使用相同的一套代碼和數(shù)據(jù)格式
3.2 不同點(diǎn)
- 設(shè)計目的不同： checkpoint是作為Flink作業(yè)容錯機(jī)制存在的具垫，用于作業(yè)潛在失敗時的恢復(fù)侈离，savepoint是作為需要作業(yè)重啟時（比如：Flink版本升級、作業(yè)拓?fù)涓淖凅莶希⑿卸刃薷呢阅耄鳂I(yè)A/B測試等場景）保存狀態(tài)并恢復(fù)的一種機(jī)制铺坞。
- 生命周期不同： checkpoint的生命周期由flink來管理，flink負(fù)責(zé)checkpoint的創(chuàng)建洲胖、維護(hù)和釋放康震，過程中沒有與用戶交互。savepoint就不同了宾濒，它是由用戶來創(chuàng)建腿短、維護(hù)和刪除的，savepoint的是事先規(guī)劃好的绘梦、手動備份并用于恢復(fù)橘忱。
- 具體實(shí)現(xiàn)不同： checkpoint作為用于恢復(fù)，需要定期觸發(fā)并保存狀態(tài)的機(jī)制卸奉。實(shí)現(xiàn)上需要滿足兩點(diǎn)：1）創(chuàng)建時盡量輕量級钝诚，2）恢復(fù)時越快越好。savepoint在創(chuàng)建和恢復(fù)時比checkpoint更重一些榄棵，更偏重于便捷性及對作業(yè)前述改動的支持凝颇。
