摘要：本文由社區(qū)志愿者佳偉整理，內(nèi)容來源自 Apache Flink Committer、阿里巴巴高級開發(fā)工程師唐云(茶干) 在 5 月 22 日北京站 Flink Meetup 分享的 《State backend Flink – 1.13 優(yōu)化及生產(chǎn)實踐分享》。內(nèi)容包括：

1. 鳥瞰 Flink 1.13 state-backend 變化

2. RocksDB state-backend 內(nèi)存管控優(yōu)化
   

3. Flink state-backend 模塊發(fā)展規(guī)劃
   

   
   

Tips：點擊文末「閱讀原文」即可查看更多技術(shù)干貨～

 GitHub 地址 

歡迎大家給 Flink 點贊送 star~

一、鳥瞰 Flink 1.13 state-backend 變化

1. State 訪問的性能監(jiān)控

首先，F(xiàn)link 1.13 中引入了 State 訪問的性能監(jiān)控，即 latency trackig state。

通過對每次訪問前后的 System#nanoTime 求差，得到 state 訪問延遲值(latency)。此功能不局限于 State Backend 的類型，自定義實現(xiàn)的 State Backend 也可以復(fù)用此功能。State 訪問的性能監(jiān)控會產(chǎn)生一定的性能影響，所以，默認每 100 次做一次取樣(sample)。上圖即監(jiān)控結(jié)果展示。

State 訪問的性能監(jiān)控開啟后，對不同的 State Backend 性能損失影響不同：

• 對于 RocksDB State Backend，性能損失大概在 1% 左右；

• 而對于 Heap State Backend，性能損失最多可達 10%。

上圖所示是三個相關(guān)的配置項，默認情況下此功能是關(guān)閉的，需通過指定參數(shù) state.backend.latency-track.keyed-state-enabled=true 來手動開啟。

2. 統(tǒng)一的 Savepoint 格式

Flink 1.1.3 之后，Savepoint 支持切換 State Backend，極大提升了系統(tǒng)應(yīng)用性。創(chuàng)建 Savepoint 后，可修改作業(yè)拓撲中 State Backend 的類型，如從 RocksDB 切換成 Heap，或從 Heap 切換成 RocksDB，但切換僅限于 Savepoint。Checkpoint 所存儲的文件格式與 State Backend 類型相關(guān)，而非通用格式，Checkpoint 目前暫不支持該功能。

3. 更清晰的 API

還有一個比較重要的改動，就是關(guān)于概念上的清晰化。Flink 1.13 中將狀態(tài)和檢查點兩者區(qū)分開來。

在 Flink 中，State Backend 有兩個功能：

1. 提供狀態(tài)的訪問、查詢；

2. 如果開啟了 Checkpoint，會周期向遠程的 Durable storage 上傳數(shù)據(jù)和返回元數(shù)據(jù) (meta) 給 Job Manager (以下簡稱 JM)。

在之前的 Flink 版本中，以上兩個功能是混在一起的，即把狀態(tài)存儲和檢查點的創(chuàng)建概念籠統(tǒng)的混在一起，導(dǎo)致初學(xué)者對此部分感覺很混亂，很難理解。

目前，State Backend 的種類如上圖所示，由于概念的混亂，導(dǎo)致之前的寫法中，RocksDB State Backend 中是可以嵌入 Memory State Backend 或 Heap State Backend 的。實際上，RocksDB 里面嵌入的 State Backend，描述的是其內(nèi)部 Checkpoint 數(shù)據(jù)傳輸方向。

對于 Memory State Backend，在原始構(gòu)建下，未指定任何的 filepath。且在不開啟 HA 的模式下，會將所有 Checkpoint 數(shù)據(jù)返回給 JM。當 Memory State Backend 指定 filepath，滿足上傳條件時，Checkpoint 數(shù)據(jù)直接上傳到指定 filepath 下，數(shù)據(jù)內(nèi)容不會返回給 JM。

對于 Fs State Backend，數(shù)據(jù)會直接上傳到所定義的 filepath 下。

當然，大家線上用的最多的還是 RocksDB State Backend 搭配上一個遠程 fs 地址，舊的寫法對于使用 Flink 的用戶來說，容易造成狀態(tài)和檢查點理解混亂。

Flink 1.13 中兩個概念被拆開：

1. 其中，State Backend 的概念變窄，只描述狀態(tài)訪問和存儲；

   
   

2. 另外一個概念是 Checkpoint storage，描述的是 Checkpoint 行為，如 Checkpoint 數(shù)據(jù)是發(fā)回給 JM 內(nèi)存還是上傳到遠程。所以，相對應(yīng)的配置項也被拆開 。

當前不僅需要指定 State Backend ，還需要指定 Checkpoint Storage。以下就是新老接口的對應(yīng)關(guān)系：

當然，雖然舊接口目前仍然保存，但還是推薦大家使用新接口，向新方式遷移，從概念上也更清晰一些。

4. RocksDB partitioned Index & filter

Flink1.13 中對 RocksDB 增加了分區(qū)索引功能。如上圖所示，RocksDB Block Cache 中存儲的數(shù)據(jù)包含三部分：

1. Data Block (真實數(shù)據(jù))

   
   

2. Index Block (每條數(shù)據(jù)的索引)

   
   

3. Filter Block (對文件的 Bloom Filter)

可以通過方塊大小明顯看出塊大小，Index 和 Filter 是明顯大于 Data 的。以 256M SSD 文件為例，Index Block 大概是 0.5M，F(xiàn)ilter Block 大概是 5M，Data Block ze 則默認是 4KB。當 Cache Block 是幾百 MB 的時，如果文件數(shù)特別多，Index 和 Filter 不斷的替出換入，性能會非常差，尤其是在默認開啟了內(nèi)存管控后。比較明顯的現(xiàn)象是，IO 特別頻繁，性能始終上不去。

Flink1.13 中，復(fù)用了 RocksDB 的 partitioned Index & filter 功能，簡單來說就是對 RocksDB 的 partitioned Index 做了多級索引。也就是將內(nèi)存中的最上層常駐，下層根據(jù)需要再 load 回來，這樣就大大降低了數(shù)據(jù) Swap 競爭。線上測試中，相對于內(nèi)存比較小的場景中，性能提升 10 倍左右。所以，如果在內(nèi)存管控下 Rocksdb 性能不如預(yù)期的話，這也能成為一個性能優(yōu)化點。

目前共有兩個參數(shù)可控制這個功能：

1. state.backend.rocksdb.memory.partitioned-index-filters:true (默認 false)

   
   

2. state.backend.rocksdb.block.metadata-blocksize (多級索引內(nèi)存配置)

5. 默認行為變化

Flink1.13 中，默認行為發(fā)生如上圖所示變化。

• 不再支持 state.backend.async 配置項，所有的 Checkpoint 均是異步的 (同步 Checkpoint 場景很少，已去除)；

• state.backend.rocksdb.checkpoint.transfer.thread.num 默認值增大到 4 RocksDB 增量 Checkpoint 時，4 個線程多線程上傳文件 RocksDB從增量 Checkpoint 恢復(fù)數(shù)據(jù)時，采用 4 個線程多線程下載。

當然，性能提升的同時，對 HDFS 底層壓力更大些，如果升級后 HDFS 不穩(wěn)定，可考慮是否與此處相關(guān)。

二、RocksDB state-backend 內(nèi)存管控優(yōu)化

Flink 1.10 開始做 state-backend 內(nèi)存優(yōu)化，在之后的每個版本中都有相關(guān)改進。

對 RocksDB State Backend 做內(nèi)存管控的最基本原因在于 Flink state 與 RocksDB 的 Column Family (獨立內(nèi)存) 一一對應(yīng)。

在 Flink 1.10 之前，如果聲明兩個 state，會各自享用自己的 Write Buffer 和 Cache 內(nèi)存，F(xiàn)link 并沒有對一個 operator 中的 state 數(shù)量限制，理論上用戶可以設(shè)置幾千個、幾萬個 state，可能導(dǎo)致容器內(nèi)存撐爆。另外，F(xiàn)link 在 slot-sharing 機制下，一個 slot 內(nèi)可以存在多個包含 keyed state 的 operator，也很難保證 state 個數(shù)不超。

多個 RocksDB 會有多個 Write Buffer Manager 。如上圖所示，以單個 Write Buffer Manager 為例，它將自己的內(nèi)存 reserve 到 Block Cache 中，根據(jù)自己的內(nèi)存管控邏輯來實現(xiàn)記賬，Block Cache 內(nèi)有 LRU Handle，超出預(yù)算時，會被踢出。

上圖提到的 arena block ，是 Write Buffer 最小內(nèi)存分配單元，默認是 Write buffer 默認配置的 1/8，內(nèi)存默認為 8MB。但在極端情況下，磁盤上會出現(xiàn)小文件過多的現(xiàn)象，導(dǎo)致性能非常差。如當整體內(nèi)存分配過小時，Write Buffer 所管控的內(nèi)存數(shù)量也就會比較少，剛開始申請內(nèi)存時，默認申請 8MB 內(nèi)存，當已用內(nèi)存達到總內(nèi)存的 7/8 時，會被置為 Immutable (置為不可變），之后這部分數(shù)據(jù)被替出到磁盤上。

如果單個 arena block 內(nèi)存占比過大，可能會出現(xiàn)臨界 arena block 只寫了幾 KB，但觸發(fā)了 Write Buffer 的內(nèi)存行為管控，將 arena block 置為了 Immutable，之后的數(shù)據(jù)就會被刷出去，產(chǎn)生小文件，導(dǎo)致性能非常差。對于 LSM DB 來說，提前 flush，對讀放大性能產(chǎn)生很大影響，Write Buffer 無法緩存更多讀請求。

我們引入對 arena block 大小有強校驗，當 arena block 大小不合適時，會打印 Warning 級別日志，認為當前需要對 arena block 大小作出相應(yīng)調(diào)整。即需要降低 arena block 大小，從而解決數(shù)據(jù)提前被 flush 的問題，進而提升性能。

RocksDB Block Cache 為了提高并發(fā)性能，將 arena block 分成了若干個分片 (shards)。實質(zhì)上是 Write Buffer Manager 在做 reserve 時，將 arena block 拆成了若干個 dummy entry，實際上只做了記賬，會占據(jù) block cache 的邏輯容量。目前 Flink 使用的 RocksDB 版本中，shards 默認是 1MB，可能會有 shards 的數(shù)據(jù)超過預(yù)算的風(fēng)險。后來的 RocksDB 高版本中，將 1MB 調(diào)成了 256KB 來解決這個風(fēng)險。由于 Flink 1.13 中沒有對 RocksDB 版本升級，所以這個問題依然存在。此外，F(xiàn)link 1.13 中，沒有將 RocksDB Block Cache 內(nèi)存管控設(shè)置成嚴格模式 (Strict Mode)。

目前社區(qū)用的 RocksDB 的版本是 5.17.2，與 RocksDB 社區(qū)最新的 6.17+ 版本，相差大概一兩千個 commit。社區(qū)在嘗試升級 RocksDB 版本時，發(fā)現(xiàn)高版本有一些性能回退，即使盡力解決，也只是解決了其中一部分，在部分訪問接口下，還是有大約不到 10% 的性能下降。所以，F(xiàn)link 1.13 決定暫不升級 RocksDB 版本，社區(qū)預(yù)計會在 Flink 1.14 中做相應(yīng)升級，引入 RocksDB 一些新的 future，借此彌補目前已知的 10% 性能回退的 Gap。

綜上各種問題，RocksDB 內(nèi)存管控不完善，加上 Writer Buffer 對 Data Block 不嚴格的管控，在理論上還是存在一定小幾率內(nèi)存超用的。但就目前來看，整體還是比較穩(wěn)定，超用的部分不會太多。如果想手動多分一部分內(nèi)存給 RocksDB 來防止超用，預(yù)防在云原生的環(huán)境因 OOM 被 K8S kill，可手動將 JVM OverHead 內(nèi)存調(diào)大，如上圖所示。

之所以不調(diào)大 Task Off-Heap，是由于目前 Task Off-Heap 是和 Direct Memeory 混在一起的，即使調(diào)大整體，也并不一定會分給 RocksDB 來做 Buffer，所以我們推薦通過調(diào)整 JVM OverHead 來解決內(nèi)存超用的問題。同理，如果 Flink 中用到其他相關(guān)庫，遇到相似問題，也可以嘗試將 JVM OverHead 調(diào)大來解決。如果想查明內(nèi)存泄漏原因，也可以結(jié)合相應(yīng) jemalloc + jeprof 等分析工具排查解決。

三、 Flink state-backend 模塊發(fā)展規(guī)劃

在 Flink 的未來規(guī)劃里，基于 Changelog 的更快速更通用的增量 Checkpoint 機制，正在開發(fā)中。而目前只有 RocksDB 支持增量 Checkpoint。    

對于 Changelog，在 Apache Kafka 和 Apache Pulsar 中都有這個概念。

Changelog 的引入，是 Flink 作為流式計算系統(tǒng)，對傳統(tǒng)消息中間件的借鑒。即在數(shù)據(jù)上傳同時，做一個 proxy，將數(shù)據(jù)定期寫到外部的 log 里，每次做 Checkpoint 時不需要等數(shù)據(jù)上傳，進而使 Checkpoint 的時間更加可控。

Flink 1.13 已經(jīng)實現(xiàn)了 proxy 代理層，實際的邏輯層還沒有實現(xiàn)，在 Flink 1.14 中會做具體實現(xiàn)，包括相關(guān) log 清理邏輯。希望在 Flink 1.14 中對狀態(tài)和檢查點性能有更好的提升，尤其是目前二階段提交依賴于 Checkpoint commit，Changelog State Backend 的引入，預(yù)計在 Flink 1.14 可以盡快解決相關(guān)痛點。