摘要：本文整理自 bilibili 資深開(kāi)發(fā)工程師張楊，在 Flink Forward Asia 2022 核心技術(shù)專場(chǎng)的分享。本篇內(nèi)容主要分為四個(gè)部分：

1. 相關(guān)背景
2. state 壓縮優(yōu)化
3. Remote state 探索
4. 未來(lái)規(guī)劃

點(diǎn)擊查看原文視頻 & 演講PPT文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-672068.html

一、相關(guān)背景

1.1 業(yè)務(wù)概況

• 從業(yè)務(wù)規(guī)模來(lái)講，B 站目前大約是 4000+的 Flink 任務(wù)，其中 95%是 SQL 類型。
• 從部署模式來(lái)講，B 站有 80%的部署是 on yarn application 部署，我們的 yarn 集群和離線的 yarn 是分開(kāi)的，是實(shí)時(shí)專用的 yarn 集群。剩下的 20%作業(yè)，為了響應(yīng)公司降本增效的號(hào)召，目前是在線集群混部。這個(gè)方案的采用主要是從成本考慮，目前在使用 yarn on docker。

從 state 使用情況來(lái)講，平臺(tái)默認(rèn)開(kāi)啟的全部是 rocksdb statebackend，大概有 50%的任務(wù)都是帶狀態(tài)的，其中有上百個(gè)任務(wù)的狀態(tài)超過(guò)了 500GB，這種任務(wù)我們稱之為超大狀態(tài)的 state 任務(wù)。

1.2 statebackend 痛點(diǎn)

主要痛點(diǎn)有 3 個(gè)：

• cpu 抖動(dòng)大。rocksdb 的原理決定了它一定會(huì)有一個(gè) compaction 的操作。目前，在小狀態(tài)下這種操作對(duì)任務(wù)本身沒(méi)什么影響，但是在大狀態(tài)下，尤其是超過(guò) 500GB 左右，compaction 會(huì)造成 cpu 抖動(dòng)大，峰值資源要求高。尤其是在 5、10、15 分鐘這種 checkpoint 的整數(shù)被間隔的情況下，整個(gè) cpu 的起伏會(huì)比較高。對(duì)于一個(gè)任務(wù)來(lái)說(shuō)，它的峰值資源要求會(huì)比較高。如果我們的資源滿足不了峰值資源的需求，任務(wù)就會(huì)觸發(fā)一陣陣的堆積和消積的情況。

• 恢復(fù)時(shí)間長(zhǎng)。因?yàn)?rocksdb 是一個(gè)嵌入式的 DB 存儲(chǔ)，每次重啟或者任務(wù)的 rescale 狀態(tài)恢復(fù)時(shí)間長(zhǎng)。因?yàn)樗押艽蟮奈募螺d下來(lái)，然后再在本地進(jìn)行一些重新的 reload，這就需要一些大量的掃描工作，所以會(huì)導(dǎo)致整個(gè)任務(wù)的狀態(tài)恢復(fù)時(shí)間比較長(zhǎng)。觀察下來(lái)，正常來(lái)說(shuō)大任務(wù)基本需要 5 分鐘左右才能恢復(fù)平穩(wěn)的狀態(tài)。

• rocksdb 非常依賴本地磁盤(pán)。rocksdb 是嵌入式的 DB，所以會(huì)非常依賴本地 io 的能力。由于今年公司有降本增效的策略，我們也會(huì)在線推薦一些混部，因?yàn)?Flink 的帶狀態(tài)任務(wù)有對(duì)本地 io 依賴的特點(diǎn)，對(duì)于混部來(lái)講是沒(méi)法用的，這就導(dǎo)致我們使用混部的資源是非常有限的。

主要通過(guò)兩個(gè)方面解決這些痛點(diǎn):

• 第一個(gè)是 state 壓縮。出于對(duì)如何降低 compaction 開(kāi)銷的思考，對(duì)任務(wù)進(jìn)行了一些 cpu 火焰圖的分析。通過(guò)分析發(fā)現(xiàn)，整個(gè) cpu 的消耗主要在 2 個(gè)方面，一個(gè)是 compaction，一個(gè)是壓縮數(shù)據(jù)的換進(jìn)換出，這就會(huì)帶來(lái)大量的壓縮和解壓縮的工作。所以我們?cè)趬嚎s層做了一些工作。
• 第二個(gè)是基于云原生的部署。由于本地沒(méi)有 io 能力，我們參考了一些云原生的方案，把整個(gè)本地的 rocksdb backend 進(jìn)行了 Remote state 的升級(jí)。這相當(dāng)于上面的一個(gè)存算分離，然后遠(yuǎn)程進(jìn)行了服務(wù)化的操作。這樣就可以實(shí)現(xiàn)即連即用，不需要進(jìn)行 reload 的操作，整個(gè) state 是基于遠(yuǎn)程化進(jìn)行的。

二、state 壓縮優(yōu)化

2.1 業(yè)務(wù)場(chǎng)景

從業(yè)務(wù)場(chǎng)景角度來(lái)看，大 state 分析下來(lái)，它主要集中于模型訓(xùn)練和樣本拼接。我們的商業(yè)化和 AI 推薦部門(mén)就是這種典型的應(yīng)用場(chǎng)景。它的特點(diǎn)就是 key 數(shù)據(jù)量龐大，分布稀疏，很難命中 cache。且數(shù)據(jù) ttl 短，基本都是小時(shí)級(jí)。

如下右圖所示，整個(gè)業(yè)務(wù)的模型包含展現(xiàn)和點(diǎn)擊。通過(guò)這兩個(gè)動(dòng)作我們進(jìn)行了校驗(yàn)，然后完成計(jì)算，從而生成樣本，再灌入模型進(jìn)行訓(xùn)練。這個(gè)場(chǎng)景下它的特點(diǎn)是緩存失效就很快失效，它的 key 數(shù)量比較大且比較稀疏。

第二個(gè)點(diǎn)是數(shù)據(jù)的提交比較短，那么整個(gè)緩存的失效會(huì)比較快，這樣就會(huì)導(dǎo)致 rocksdb 在使用的時(shí)候會(huì)反復(fù)從磁盤(pán)里不斷地 reload 數(shù)據(jù)，然后進(jìn)行壓縮和解壓縮，或者中間 compaction 數(shù)據(jù)不斷地過(guò)期，導(dǎo)致 compaction 過(guò)于頻繁。

2.2 優(yōu)化思路

以上兩個(gè)問(wèn)題會(huì)導(dǎo)致整個(gè) cpu 的消耗非常高，所以我們能觀察到，在前面提及的整個(gè)任務(wù)的峰值 cpu 消耗會(huì)非常高。我們的優(yōu)化思路是通過(guò)包括對(duì)社區(qū)的一些方案和業(yè)界其他公司分享的一些文章進(jìn)行的調(diào)研，調(diào)研和優(yōu)化方案主要圍繞三個(gè)方面進(jìn)行：

• 第一個(gè)是開(kāi)啟 partitoned index filter，減少緩存競(jìng)爭(zhēng)。從火焰圖中可以看到，data block 會(huì)不斷地從 block cache 中被換進(jìn)換出，同時(shí)產(chǎn)生大量的壓縮和解壓縮工作，大量的 cpu 其實(shí)消耗在這里。

• 第二個(gè)是關(guān)閉 rocksdb 壓縮，減少緩存加載 /compaction 時(shí)候的 cpu 消耗。從火焰圖中可以看到，整個(gè) SSD 數(shù)據(jù)進(jìn)行底層生成的時(shí)候，它會(huì)不斷地進(jìn)行整個(gè)文件的壓縮和解壓縮。所以我們的思路就是如何去關(guān)閉一些 rocksdb 壓縮，減少整個(gè)緩存加載或者 cpu 的消耗。

• 第三個(gè)是支持接口層壓縮，在數(shù)據(jù) state 讀寫(xiě)前后進(jìn)行解壓縮操作，減少 state 大小。如果關(guān)閉 rocksdb 底層的壓縮，我們觀察到整個(gè)磁盤(pán)的使用量非常高。這樣的話，壓力就是從壓縮和解壓縮的 cpu 消耗變成磁盤(pán)的 io 消耗，磁盤(pán)的 io 很容易就被消耗盡了。所以我們就把整個(gè)底層的壓縮往上層進(jìn)行了一個(gè)牽引，支持接口層的壓縮。

在數(shù)據(jù)層寫(xiě)入讀寫(xiě)前后進(jìn)行壓縮和解壓縮的操作，減少底層的存儲(chǔ)大小，我們也與內(nèi)部的一些 DB 存儲(chǔ)團(tuán)隊(duì)進(jìn)行了一些溝通，他們一般建議在上層進(jìn)行壓縮，這樣會(huì)比在底層做效率更高一些。因?yàn)榈讓诱麄€(gè)是以 Block 維度進(jìn)行壓縮與解壓縮的。

2.3 整體架構(gòu)

partitoned index filter 開(kāi)啟，B 站內(nèi)部目前主要用的是 Flink 1.11 版本。當(dāng)初我們大概調(diào)研到這個(gè)參數(shù)的時(shí)候，發(fā)現(xiàn)社區(qū)本身在 1.12 版本已經(jīng)提升了這個(gè)功能，所以就直接把 1.12 版本相關(guān)的功能直接用過(guò)來(lái)了，同時(shí) rocksdb 壓縮關(guān)閉，官方文檔均有參數(shù)，這個(gè)部分成本比較低。

第二個(gè)部分是支持業(yè)務(wù)的壓縮。為了減少磁盤(pán) io 的壓力，在接口上做了一些壓縮。接口壓縮架構(gòu)，如上圖，重寫(xiě)了所有 rocksdb state 的 put/get 接口，使用的是最新的 Zstd 壓縮算法。目前使用的是 LTZ 算法，比一些常用的壓縮收益更高，cpu 的消耗比較小。

2.4 業(yè)務(wù)效果

如上業(yè)務(wù)效果對(duì)比圖，這里面包含 test 1、2、3、4，主要看看 test 0 和 test 4。test 0 是社區(qū)默認(rèn)的方案，也就是底層用是 snappy 壓縮，上層不做什么。test 4 是完全關(guān)閉了底層的壓縮，無(wú)論是上層還是底層，整個(gè)壓縮是全部關(guān)閉的，在業(yè)務(wù)層使用了最新的 STD 方法進(jìn)行壓縮。

整個(gè)觀察下來(lái)，峰值和低峰期的數(shù)據(jù)都進(jìn)行了查看，cpu 均值大概是下降到 15%左右，峰值大概下降到 25%，相信對(duì)毛刺的下降會(huì)更明顯一些。另外，在業(yè)務(wù)層做完壓縮之后，這個(gè)文件大小基本保持不變，或是只提升了一點(diǎn)點(diǎn)。這樣一來(lái)，最底層的存儲(chǔ)空間或者 io 其實(shí)完全沒(méi)有增加任何壓力。

目前來(lái)講，線上所有新增的大的任務(wù)基本都已經(jīng)全部開(kāi)啟了，老的任務(wù)也在進(jìn)行一些推進(jìn)。

最后補(bǔ)充下這個(gè)場(chǎng)景的適應(yīng)性。因?yàn)檎麄€(gè)方案是匹配到我們的推薦功能或是廣告模型拼接場(chǎng)景來(lái)做的，所有的性能分析也是在這樣的場(chǎng)景下進(jìn)行的，結(jié)論是主要雙流 join 的大 state 場(chǎng)景，小 state 場(chǎng)景下基本沒(méi)有收益，主要原因是小 state 的模式下，cache 本身的命中率非常高，換進(jìn)換出頻率較低。另外，Liststate 暫時(shí)不支持業(yè)務(wù)層壓縮，add 接口底層的特殊 merge 實(shí)現(xiàn)暫時(shí)無(wú)法做業(yè)務(wù)層壓縮。

三、Remote state 探索

3.1 業(yè)務(wù)場(chǎng)景和問(wèn)題

第一，降本增效。這是今年大部分互聯(lián)網(wǎng)公司可能都在做的事情，B 站也不列外。我們?cè)谕埔粋€(gè)離在線混部，主要是為了提升資源的使用效率。把 Flink 大數(shù)據(jù)的場(chǎng)景和在線服務(wù)的場(chǎng)景進(jìn)行資源的混部。

第二，大 state 作業(yè)重啟下載 state 慢。有些用戶反應(yīng)，對(duì)一些重啟時(shí)間比較敏感的業(yè)務(wù)方，會(huì)覺(jué)得大狀態(tài)下的重啟對(duì)業(yè)務(wù)會(huì)有一些抖動(dòng)，尤其在廣告、AI 模型推薦等方面體驗(yàn)感不好。

第三，大 state 作業(yè)重啟加載 state 慢。希望在重啟或是擴(kuò)容的時(shí)候，加載更快一些。這里包括兩個(gè)業(yè)務(wù)場(chǎng)景，一個(gè)是離在線混部，第二個(gè)是用戶的重啟體驗(yàn)。這兩個(gè)方面即使是用 rocksdb 來(lái)做都比較難。例如在線混部，因?yàn)樵诰€的機(jī)器的磁盤(pán)基本是 100-200GB 的那種本地較小的磁板，同時(shí) io 性能也不是很高，在這種情況下，將Flink大任務(wù)丟到在線混部上去，會(huì)對(duì)在線的機(jī)器負(fù)載壓力比較大。這就造成在線混部上的任務(wù)是相對(duì)有限的。

其實(shí)用戶很難區(qū)分到底哪些任務(wù)是可以丟到在線混部，哪些不能。這需要用戶通過(guò) state 來(lái)判斷。另外， state 的慢導(dǎo)致用戶體驗(yàn)比較差，主要是因?yàn)樗貑⒌臅r(shí)候強(qiáng)依賴 load 過(guò)程，尤其是當(dāng)遇到超過(guò) 500GB 的任務(wù)時(shí)候，這個(gè)體驗(yàn)感就會(huì)比較差。

3.2 優(yōu)化思路

首先是實(shí)現(xiàn) Flink state 的存算分離。如果進(jìn)行混部，本地實(shí)際上沒(méi)有磁盤(pán)，就很難給到在線的機(jī)器，在 Flink 上最直接的辦法就是存算分離。存算分離不是陌生的概念，尤其是在云上。存儲(chǔ)放在存儲(chǔ)的機(jī)器上，按照存儲(chǔ)的需求進(jìn)行機(jī)型采購(gòu)，計(jì)算同理。

其次是 state 遠(yuǎn)程化/服務(wù)化，重啟 /rescale 直接建立連接進(jìn)行讀寫(xiě)。B 站的做法是把 Flink 的 state 進(jìn)行存儲(chǔ)的分離，這樣就能解決在線的混部問(wèn)題。分離之后，就可以進(jìn)行遠(yuǎn)程化和服務(wù)化的部署，最終重啟或是其他就能直接與遠(yuǎn)程服務(wù)進(jìn)行連接。

第三是遠(yuǎn)程 state 服務(wù)需要支持 checkpoint 完整功能依賴的接口。這是正常開(kāi)發(fā)流程中所依賴的，即遠(yuǎn)程服務(wù)。需要支持一個(gè) checkpoint 的完整功能，主要對(duì)標(biāo)的就是 rocksdb 的功能。

3.3 整體架構(gòu)

上圖是 Flink state 整體的架構(gòu)，也是 statebackend 對(duì)象拓?fù)?，就是上圖右側(cè)。另外，F(xiàn)link statebackend 的核心數(shù)據(jù)或是數(shù)據(jù)存儲(chǔ)主要是由 keyed statebackend 承擔(dān)，還有一部分是由 operator state 承擔(dān)，這部分主要是存在內(nèi)存里，并全量往遠(yuǎn)程刷，一般支持的數(shù)據(jù)不會(huì)特別大。

核心的大規(guī)模數(shù)據(jù)會(huì)存在 keyed statebackend 里。keyed statebackend 主要有 heap 和 rocksdb 兩種實(shí)現(xiàn)，支持大規(guī)模額數(shù)據(jù)存儲(chǔ)。

B 站主要使用的是 rocksdb 存儲(chǔ)，主要圍繞的核心是改進(jìn) keyed statebackend，因?yàn)?operator state 主要針對(duì)的是小規(guī)模數(shù)據(jù)，它不會(huì)占用本地 io，也不會(huì)在重啟的時(shí)候產(chǎn)生大量的重復(fù)操作，速度相對(duì)來(lái)說(shuō)要快很多。

上圖是社區(qū) keyed statebackend 的分配邏輯架構(gòu)。如果想要把 keyed statebackend 做存算分離，需要了解：

• Flink 的分片邏輯，它的底層有一個(gè) keyed group 的概念。在所有狀態(tài)在任務(wù)第一次啟動(dòng)的時(shí)候，底層分片就已經(jīng)確定好了。這個(gè)分片大小就是 Flink 的最大并發(fā)度的值。這個(gè)最大并發(fā)度實(shí)際上是在任務(wù)第一次不從 checkpoint 啟動(dòng)時(shí)候算出來(lái)的。這中間如果一直從 checkpoint 啟動(dòng)，那么最大并發(fā)度永遠(yuǎn)不會(huì)變。

• Flink checkpoint rescale 能力是有限的，本質(zhì)是分片 key-group 在 subtask 之間的移動(dòng)，分片無(wú)法分裂。一旦啟動(dòng)生成切換之后，這個(gè)分片的數(shù)量因?yàn)橐呀?jīng)固定好了，后面不管怎么 rescale，本質(zhì)是分片的移動(dòng)，其實(shí)最大的變動(dòng)是后面 rescale 不能超過(guò)整個(gè)分片的數(shù)量，這個(gè)數(shù)量對(duì)應(yīng)的是分片最大的并發(fā)度。

• 多個(gè) key-group 存儲(chǔ)在相同 rocksdb 的 cf 里面，以 key-group 的 ID 作為 key 的 prefix，依賴 rocksdb 的排序特性，rescale 過(guò)程通過(guò) prefix 進(jìn)行定位遍歷提取。

上圖是 Remote state 整體架構(gòu)，主要的核心工作有三點(diǎn)：

• 替換 rocksdb 為 B 站內(nèi)部的 taishan 存儲(chǔ)。Taishan 存儲(chǔ)底層是分布式結(jié)構(gòu)，F(xiàn)link 的一些概念是可以對(duì)應(yīng)到 taishan 存儲(chǔ)上的。

• key-group 概念對(duì)應(yīng) taishan 存儲(chǔ)的 shard，cf 對(duì)應(yīng) table。key-group 的概念就可以對(duì)應(yīng)上 taishan 存儲(chǔ)，同時(shí)里面 cf 的概念也對(duì)應(yīng) taishan 存儲(chǔ)中的一個(gè) table 概念，也支持一個(gè) table 里面可以有多個(gè) key-group。

• Taishan 存儲(chǔ)支持對(duì)應(yīng) shard 的 snapshot，以及從歷史的 snapshot 進(jìn)行快速切換。比如重啟的時(shí)候可以從老的 checkpoint 進(jìn)行回復(fù)，實(shí)際上能夠調(diào)用 taishan 存儲(chǔ)這樣一套接口進(jìn)行快速 snapshot 切換，并可以指定對(duì)應(yīng)的 snapshot 并進(jìn)行加載，加載的速度非?？欤臼敲爰?jí)左右。

3.4 緩存架構(gòu)

在上文介紹的基礎(chǔ)上可以觀察到，開(kāi)發(fā)完之后實(shí)際上整個(gè)功能來(lái)說(shuō)是完全夠用的，但也發(fā)現(xiàn)了一些比較嚴(yán)重的性能問(wèn)題：

• 每次 state 操作都需要走網(wǎng)絡(luò) rpc， cpu 的消耗太高

• 網(wǎng)絡(luò) rpc 的延遲高，任務(wù)吞吐低

• cache 是一個(gè)自然而然的選擇

3.5 緩存難點(diǎn)

寫(xiě)緩存相對(duì)來(lái)講是比較簡(jiǎn)單，內(nèi)存攢批，后臺(tái)定期 checkpoint 刷出去就可以上百倍的減少寫(xiě)的 rpc。

讀緩存是比較難的。讀緩存是把場(chǎng)景進(jìn)行拆分來(lái)看：

• key 比較少的場(chǎng)景效果明顯，命中率極高，效果好，甚至達(dá)到幾百上千倍的效果提升。

• 稀疏的 key 場(chǎng)景，大量讀 null，緩存命中率低。這個(gè)場(chǎng)景是當(dāng)下比較大的難點(diǎn)。

• 周期性業(yè)務(wù)導(dǎo)致緩存定期失效，讀 null，命中率暴跌。周期性業(yè)務(wù)到周期末節(jié)點(diǎn)，就會(huì)觸發(fā)一個(gè)緩存失效，一旦跨過(guò)這個(gè)節(jié)點(diǎn)，緩存里就會(huì)變成新的數(shù)據(jù)。如果正巧遇上讀 null，命中率就會(huì)很低，甚至導(dǎo)致任務(wù)抖動(dòng)。這也是比較大的難點(diǎn)。

基于以上的問(wèn)題，優(yōu)化方案是支持 key 全量緩存配置，有效解決稀疏/緩存失效場(chǎng)景性能問(wèn)題，使內(nèi)存使用上升。但是這個(gè)優(yōu)化缺點(diǎn)也比較明顯，即對(duì)內(nèi)存的要求會(huì)高些。因?yàn)樗械?key 都要存在一個(gè)緩存里，而這個(gè)緩存就是內(nèi)存。

除了以上介紹的兩點(diǎn)難點(diǎn)，其實(shí)還有另外兩個(gè)難點(diǎn)。其一是冷啟動(dòng)，因?yàn)榫彺娌粫?huì)進(jìn)行持久化，那么當(dāng)任務(wù)重啟等狀態(tài)下，其實(shí)重啟之后緩存內(nèi)是空的，那么也就會(huì)造成重啟的時(shí)候，讀和寫(xiě)的請(qǐng)求量都很大，會(huì)給性能造成壓力。

另外，全量 key 的緩存，啟動(dòng)的時(shí)候也沒(méi)法立即使用，需要等一個(gè)ttl周期，這樣才能可以保證后面的全量 key 緩存是真正的全量 key。

3.6 off-heap 緩存模型圖

上圖是 off-heap 緩存模型，我們使用的是 Facebook 的 OHC（off-heap cache）?？梢杂^察到，基于 ttl 的主動(dòng)淘汰，它帶來(lái)的性能消耗相對(duì)來(lái)講會(huì)比較大。但是如果完全關(guān)掉基于ttl的淘汰機(jī)制，在某些特殊場(chǎng)景，基于 io 淘汰機(jī)制，也會(huì)消耗比較大的空間。因?yàn)?io 淘汰的時(shí)候會(huì)進(jìn)行判斷哪個(gè) key 可以真的被淘汰。如果完全通過(guò) io 淘汰，在有些場(chǎng)景下面可能判定了很多 key，這些 key 都無(wú)法滿足一個(gè) io 的淘汰條件，那么這個(gè)淘汰流程就會(huì)很長(zhǎng)，會(huì)導(dǎo)致接口卡的時(shí)間會(huì)比較久。所以在這里 B 站進(jìn)行了調(diào)優(yōu)，既保留了一個(gè)主動(dòng)淘汰的思路，也依賴 io 這樣一個(gè)淘汰機(jī)制。

目前來(lái)講，我們已經(jīng)完成針對(duì)部分任務(wù)的灰度，灰度任務(wù)的計(jì)算資源消耗降低了 30%左右。遠(yuǎn)程 state 的服務(wù)消耗暫時(shí)沒(méi)有統(tǒng)計(jì)，粗略看整體的資源應(yīng)該是持平或是只有略微一點(diǎn)升高。離在線混部以及快速 rescale 的意義是巨大的。Flink 側(cè)可以實(shí)現(xiàn)存算分離的效果，同時(shí)，這也能夠做快速的重啟而不需要進(jìn)行狀態(tài)的重新下載、重新加載或是 reload 的過(guò)程。

四、未來(lái)規(guī)劃

點(diǎn)擊查看原文視頻 & 演講PPT