作者介紹

邱從賢（山智），Apache Flink Contributor，中南大學碩士，2018 年加入阿里巴巴計算平臺事業(yè)部，專注于 Flink 核心引擎開發(fā)，主要從事 Flink State&Checkpoint 相關研發(fā)工作。

眾所周知，Flink 是當前最為廣泛使用的計算引擎之一，它使用 checkpoint 機制進行容錯處理 [1]，checkpoint 會將狀態(tài)快照備份到分布式存儲系統(tǒng)，供后續(xù)恢復使用。在 Alibaba 內(nèi)部，我們使用的存儲主要是 HDFS，當同一個集群的 Job 到達一定數(shù)量后，會對 HDFS 造成非常大的壓力，本文將介紹一種大幅度降低 HDFS 壓力的方法——小文件合并。

背景

不管使用 FsStateBackend、RocksDBStateBackend 還是 NiagaraStateBackend，Flink 在進行 checkpoint 的時候，TM 會將狀態(tài)快照寫到分布式文件系統(tǒng)中，然后將文件句柄發(fā)給 JM，JM 完成全局 checkpoint 快照的存儲，如下圖所示：

對于全量 checkpoint 來說，TM 將每個 checkpoint 內(nèi)部的數(shù)據(jù)都寫到同一個文件，而對于 RocksDBStateBackend/NiagaraStateBackend 的增量 checkpoint [2] 來說，則會將每個 sst 文件寫到一個分布式系統(tǒng)的文件內(nèi)。當作業(yè)量很大，且作業(yè)的并發(fā)很大時，則會對底層 HDFS 形成非常大的壓力：1）大量的 RPC 請求會影響 RPC 的響應時間（如下圖所示）；2）大量文件對 NameNode 內(nèi)存造成很大壓力。

?

在 Flink 中曾經(jīng)嘗試使用 ByteStreamStateHandle 來解決小文件多的問題 [3]，將小于一定閾值的 state 直接發(fā)送到 JM，由 JM 統(tǒng)一寫到分布式文件中，從而避免在 TM 端生成小文件。但是這個方案有一定的局限性，閾值設置太小，還會有很多小文件生成，閾值設置太大，則會導致 JM 內(nèi)存消耗太多有 OOM 的風險。

一、小文件合并方案

針對上面的問題我們提出一種解決方案——小文件合并。

在原來的實現(xiàn)中，每個 sst 文件會打開一個 CheckpointOutputStream，每個 CheckpointOutputStream 對應一個 FSDataOutputStream，將本地文件寫往一個分布式文件，然后關閉 FSDataOutputStream，生成一個 StateHandle。如下圖所示：

小文件合并則會重用打開的 FSDataOutputStream，直至文件大小達到預設的閾值為止，換句話說多個 sst 文件會重用同一個 DFS 上的文件，每個 sst 文件占用 DFS 文件中的一部分，最終多個 StateHandle 共用一個物理文件，如下圖所示：

在接下來的章節(jié)中我們會描述實現(xiàn)的細節(jié)，其中需要重點考慮的地方包括：

1）并發(fā) checkpoint 的支持

Flink 天生支持并發(fā) checkpoint，小文件合并方案則會將多個文件寫往同一個分布式存儲文件中，如果考慮不當，數(shù)據(jù)會寫串或者損壞，因此我們需要有一種機制保證該方案的正確性，詳細描述參考 2.1 節(jié)。

2）防止誤刪文件

我們使用引用計數(shù)來記錄文件的使用情況，僅通過文件引用計數(shù)是否降為 0 進行判斷刪除，則可能誤刪文件，如何保證文件不會被錯誤刪除，我們將會在 2.2 節(jié)進行闡述。

3）降低空間放大

使用小文件合并之后，只要文件中還有一個 statehandle 被使用，整個分布式文件就不能被刪除，因此會占用更多的空間，我們在 2.3 節(jié)描述了解決該問題的詳細方案。

4）異常處理

我們將在 2.4 節(jié)闡述如何處理異常情況，包括 JM 異常和 TM 異常的情況。

2.5 節(jié)中會詳細描述在 Checkpoint 被取消或者失敗后，如何取消 TM 端的 Snapshot，如果不取消 TM 端的 Snapshot，則會導致 TM 端實際運行的 Snapshot 比正常的多。

在第 3 節(jié)中闡述了小文件合并方案與現(xiàn)有方案的兼容性；第 4 節(jié)則會描述小文件合并方案的優(yōu)勢和不足；最后在第 5 節(jié)我們展示在生產(chǎn)環(huán)境下取得的效果。

二、設計實現(xiàn)

本節(jié)中我們會詳細描述整個小文件合并的細節(jié)，以及其中的設計要點。

這里我們大致回憶一下 TM 端 Snapshot 的過程：

• TM 端 barrier 對齊
• TM Snapshot 同步操作
• TM Snapshot 異步操作

其中上傳 sst 文件到分布式存儲系統(tǒng)在上面的第三步，同一個 checkpoint 內(nèi)的文件順序上傳，多個 checkpoint 的文件上傳可能同時進行。

1、并發(fā) checkpoint 支持

Flink 天生支持并發(fā) checkpoint，因此小文件合并方案也需要能夠支持并發(fā) checkpoint，如果不同 checkpoint 的 sst 文件同時寫往一個分布式文件，則會導致文件內(nèi)容損壞，后續(xù)無法從該文件進行 restore。

在 FLINK-11937[4] 的提案中，我們會將每個 checkpoint 的 state 文件寫到同一個 HDFS 文件，不同 checkpoint 的 state 寫到不同的 HDFS 文件 -- 換句話說，HDFS 文件不跨 Checkpoint 共用，從而避免了多個客戶端同時寫入同一個文件的情況。

后續(xù)我們會繼續(xù)推進跨 Checkpoint 共用文件的方案，當然在跨 Checkpoint 共用文件的方案中，并行的 Checkpoint 也會寫往不同的 HDFS 文件。

2、防止誤刪文件

復用底層文件之后，我們使用引用計數(shù)追蹤文件的使用情況，在文件引用數(shù)降為 0 的情況下刪除文件。但是在某些情況下，文件引用數(shù)為 0 的時候，并不代表文件不會被繼續(xù)使用，可能導致文件誤刪。下面我們會詳細描述開啟并發(fā) checkpoint 后可能導致文件誤刪的情況，以及解決方案。

以下圖為例，maxConcurrentlyCheckpoint = 2：

上圖中共有 3 個 checkpoint，其中 chk-1 已經(jīng)完成，chk-2 和 chk-3 都基于 chk-1 進行，chk-2 在 chk-3 前完成，chk-3 在注冊 4.sst 的時候發(fā)現(xiàn)，發(fā)現(xiàn) 4.sst 在 chk-2 中已經(jīng)注冊過，會重用 chk-2 中 4.sst 對應的 stateHandle，然后取消 chk-3 中的 4.sst 的注冊，并且刪除 stateHandle，在處理完 chk-3 中 4.sst 之后，該 stateHandle 對應的分布式文件的引用計數(shù)為 0，如果我們這個時候刪除分布式文件，則會同時刪除 5.sst 對應的內(nèi)容，導致后續(xù)無法從 chk-3 恢復。

這里的問題是如何在 stateHandle 對應的分布式文件引用計數(shù)降為 0 的時候正確判斷是否還會繼續(xù)引用該文件，因此在整個 checkpoint 完成處理之后再判斷某個分布式文件能否刪除，如果真?zhèn)€ checkpoint 完成發(fā)現(xiàn)文件沒有被引用，則可以安全刪除，否則不進行刪除。

3、降低空間放大

使用小文件合并方案后，每個 sst 文件對應分布式文件中的一個 segment，如下圖所示：

文件僅能在所有 segment 都不再使用時進行刪除，上圖中有 4 個 segment，僅 segment-4 被使用，但是整個文件都不能刪除，其中 segment[1-3] 的空間被浪費掉了，從實際生產(chǎn)環(huán)境中的數(shù)據(jù)可知，整體的空間放大率（實際占用的空間 / 真實有用的空間）在 1.3 - 1.6 之間。

為了解決空間放大的問題，在 TM 端起異步線程對放大率超過閾值的文件進行壓縮。而且僅對已經(jīng)關閉的文件進行壓縮。

整個壓縮的流程如下所示：

• 計算每個文件的放大率
• 如果放大率較小則直接跳到步驟 7
• 如果文件 A 的放大率超過閾值，則生成一個對應的新文件 A‘（如果這個過程中創(chuàng)建文件失敗，則由 TM 負責清理工作）
• 記錄 A 與 A’ 的映射關系
• 在下一次 checkpoint X 往 JM 發(fā)送落在文件 A 中的 StateHandle 時，則使用 A` 中的信息生成一個新的 StateHandle 發(fā)送給 JM
• checkpoint X 完成后，我們增加 A‘ 的引用計數(shù)，減少 A 的引用計數(shù)，在引用計數(shù)降為 0 后將文件 A 刪除（如果 JM 增加了 A’ 的引用，然后出現(xiàn)異常，則會從上次成功的 checkpoint 重新構建整個引用計數(shù)器）
• 文件壓縮完成

4、異常情況處理

在 checkpoint 的過程中，主要有兩種異常：JM 異常和 TM 異常，我們將分情況闡述。

1）JM 異常

JM 端主要記錄 StateHandle 以及文件的引用計數(shù)，引用計數(shù)相關數(shù)據(jù)不需要持久化到外存中，因此不需要特殊的處理，也不需要考慮 transaction 等相關操作，如果 JM 發(fā)送 failover，則可以直接從最近一次 complete checkpoint 恢復，并重建引用計數(shù)即可。

2）TM 異常

TM 異常可以分為兩種：1）該文件在之前 checkpoint 中已經(jīng)匯報過給 JM；2）文件尚未匯報過給 JM，我們會分情況闡述。

①文件已經(jīng)匯報過給 JM

文件匯報過給 JM，因此在 JM 端有文件的引用計數(shù)，文件的刪除由 JM 控制，當文件的引用計數(shù)變?yōu)?0 之后，JM 將刪除該文件。

②文件尚未匯報給 JM

該文件暫時尚未匯報過給 JM，該文件不再被使用，也不會被 JM 感知，成為孤兒文件。這種情況暫時有外圍工具統(tǒng)一進行清理。

5、取消 TM 端 snapshot

像前面章節(jié)所說，我們需要在 checkpoint 超時 / 失敗時，取消 TM 端的 snapshot，而 Flink 則沒有相應的通知機制，現(xiàn)在 FLINK-8871[5] 在追蹤相應的優(yōu)化，我們在內(nèi)部增加了相關實現(xiàn)，當 checkpoint 失敗時會發(fā)送 RPC 數(shù)據(jù)給 TM，TM 端接受到相應的 RPC 消息后，會取消相應的 snapshot。

三、兼容性

小文件合并功能支持從之前的版本無縫遷移過來。從之前的 checkpoint restore 的的步驟如下：

• 每個 TM 分到自己需要 restore 的 state handle
• TM 從遠程下載 state handle 對應的數(shù)據(jù)
• 從本地進行恢復

小文件合并主要影響的是第 2 步，從遠程下載對應數(shù)據(jù)的時候?qū)Σ煌?StateHandle 進行適配，因此不影響整體的兼容性。

四、優(yōu)勢和不足

優(yōu)勢：大幅度降低 HDFS 的壓力：包括 RPC 壓力以及 NameNode 內(nèi)存的壓力。

不足：不支持 State 多線程上傳的功能（State 上傳暫時不是 checkpoint 的瓶頸）。

五、線上環(huán)境的結果

在該方案上線后，對 Namenode 的壓力大幅降低，下面的截圖來自線上生產(chǎn)集群，從數(shù)據(jù)來看，文件創(chuàng)建和關閉的 RPC 有明顯下降，RPC 的響應時間也有大幅度降低，確保順利度過雙十一。

參考資料

• [1] https://ci.apache.org/projects/flink/flink-docs-stable/ops/state/checkpoints.html
• [2] https://flink.apache.org/features/2018/01/30/incremental-checkpointing.html
• [3] https://www.slideshare.net/dataArtisans/stephan-ewen-experiences-running-flink-at-very-large-scale
• [4] https://issues.apache.org/jira/browse/FLINK-11937
• [5] https://issues.apache.org/jira/browse/FLINK-8871

《新程序員》：云原生和全面數(shù)字化實踐50位技術專家共同創(chuàng)作，文字、視頻、音頻交互閱讀

總結

以上是生活随笔為你收集整理的阿里大规模应用Flink踩过的坑：如何大幅降低HDFS压力？的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。