如何在Apache Flink中使用RocksDB狀態(tài)后端，針對這個問題，這篇文章詳細(xì)介紹了相對應(yīng)的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

創(chuàng)新互聯(lián)長期為上千多家客戶提供的網(wǎng)站建設(shè)服務(wù)，團隊從業(yè)經(jīng)驗10年，關(guān)注不同地域、不同群體，并針對不同對象提供差異化的產(chǎn)品和服務(wù)；打造開放共贏平臺，與合作伙伴共同營造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為河北企業(yè)提供專業(yè)的做網(wǎng)站、成都做網(wǎng)站，河北網(wǎng)站改版等技術(shù)服務(wù)。擁有十余年豐富建站經(jīng)驗和眾多成功案例,為您定制開發(fā)。

流處理應(yīng)用程序通常是有狀態(tài)的，“記住”已處理事件中的信息，并使用它來影響進一步的事件處理。在Flink中，記住的信息，即狀態(tài)，被本地存儲在配置的狀態(tài)后端中。為了防止發(fā)生故障時丟失數(shù)據(jù)，狀態(tài)后端會定期將其內(nèi)容的快照持久化到預(yù)先配置的持久存儲中。RocksDB狀態(tài)后端（即RocksDBStateBackend）是Flink中三個內(nèi)置狀態(tài)后端之一。這篇博文將引導(dǎo)您了解使用RocksDB管理應(yīng)用程序狀態(tài)的好處，解釋何時以及如何使用它，并澄清一些常見的誤解。

Flink中的狀態(tài)

為了更好地理解Flink中的狀態(tài)和狀態(tài)后端，區(qū)分飛行狀態(tài)(in-flight state)和狀態(tài)快照(state snapshots)是很重要的。飛行狀態(tài)，也稱為工作狀態(tài)，是Flink作業(yè)正在處理的狀態(tài)。它總是本地存儲在內(nèi)存中（有可能溢出到磁盤），并且在作業(yè)失敗時可能會丟失，而不會影響作業(yè)的可恢復(fù)性。狀態(tài)快照，即檢查點和保存點，存儲在遠程持久存儲器中，用于在作業(yè)失敗時恢復(fù)本地狀態(tài)。生產(chǎn)部署的適當(dāng)狀態(tài)后端取決于可伸縮性、吞吐量和延遲要求。

什么是RocksDB?

認(rèn)為RocksDB是一個分布式數(shù)據(jù)庫，需要在集群上運行并由專門的管理員管理，這是一個常見的誤解。RocksDB是一個用于快速存儲的可嵌入持久鍵值存儲。它通過Java本機接口（JNI）與Flink交互。下圖顯示了RocksDB在Flink集群節(jié)點中的位置。以下各節(jié)將詳細(xì)說明。

Flink中的RocksDB

將RocksDB用作狀態(tài)后端所需的一切都捆綁在Apache Flink發(fā)行版中，包括本機共享庫：

$ jar -tvf lib/flink-dist_2.12-1.12.0.jar| grep librocksdbjni-linux64
8695334 Wed Nov 27 02:27:06 CET 2019 librocksdbjni-linux64.so

在運行時，RocksDB嵌入到TaskManager進程中。它在本機線程中運行并處理本地文件。例如，如果你的Flink集群中有一個配置了RocksDBStateBendback的作業(yè)，您將看到類似于下面的內(nèi)容，其中32513是TaskManager進程ID。

$ ps -T -p 32513 | grep -i rocksdb
32513 32633 ?        00:00:00 rocksdb:low0
32513 32634 ?        00:00:00 rocksdb:high0

注意：該命令僅適用于Linux。對于其他操作系統(tǒng)，請參閱其文檔

什么時候使用RocksDBStateBackend

除了RocksDBStateBackend，F(xiàn)link還有另外兩個內(nèi)置的狀態(tài)后端：MemoryStateBackend和FsStateBackend。它們都是基于堆的，因為運行中的狀態(tài)存儲在JVM堆中。目前，讓我們忽略MemoryStateBackend，因為它只用于本地開發(fā)和調(diào)試，不用于生產(chǎn)。

使用RocksDBStateBackend，運行中的狀態(tài)首先寫入堆外/本機內(nèi)存，然后在達到配置的閾值時刷新到本地磁盤。這意味著RocksDBStateBackend可以支持大于總配置堆容量的狀態(tài)。可以存儲在RocksDBStateBackend中的狀態(tài)量僅受整個集群中可用磁盤空間的限制。此外，由于RocksDBStateBackend不使用JVM堆來存儲運行中的狀態(tài)，因此它不受JVM垃圾收集的影響，因此具有可預(yù)測的延遲。

除了完整的、自包含的狀態(tài)快照之外，RocksDBStateBackend還支持作為性能調(diào)優(yōu)選項的增量檢查點。增量檢查點僅存儲自上次完成的檢查點以來發(fā)生的更改。與執(zhí)行完整快照相比，這大大減少了檢查點時間。RocksDBStateBendback是當(dāng)前唯一支持增量檢查點的狀態(tài)后端。

在以下情況下，RocksDB是一個不錯的選擇：

• 作業(yè)的狀態(tài)超出了本地內(nèi)存的容量（例如，長時間的窗口、大的KeyedState）；

• 你正在研究增量檢查點作為一種減少檢查點時間的方法；

• 希望有更可預(yù)測的延遲，而不受JVM垃圾回收的影響

否則，如果應(yīng)用程序的狀態(tài)很小或需要很低的延遲，則應(yīng)該考慮FsStateBackend。根據(jù)經(jīng)驗，RocksDBStateBackend比基于堆的狀態(tài)后端慢幾倍，因為它將鍵值對存儲為序列化的字節(jié)。這意味著任何狀態(tài)訪問（讀/寫）都需要經(jīng)過一個跨JNI邊界的反序列化/序列化過程，這比直接使用堆上的狀態(tài)表示更昂貴。好處是，對于相同數(shù)量的狀態(tài)，與相應(yīng)的堆上表示法相比，它的內(nèi)存占用率較低。

如何使用RocksDBStateBackend

RocksDB完全嵌入到TaskManager進程中，并完全由TaskManager進程管理。RocksDBStateBackend可以在集群級別配置為整個集群的默認(rèn)值，也可以在作業(yè)級別配置為單個作業(yè)的默認(rèn)值。作業(yè)級配置優(yōu)先于集群級配置。

集群級別

在conf/flink-conf.yaml中添加以下配置:

state.backend: rocksdb
state.backend.incremental: true
state.checkpoints.dir: hdfs:///flink-checkpoints # location to store checkpoints

作業(yè)級別

創(chuàng)建StreamExecutionEnvironment后，將以下內(nèi)容添加到作業(yè)的代碼中：

# 'env' is the created StreamExecutionEnvironment
# 'true' is to enable incremental checkpointing
env.setStateBackend(new RocksDBStateBackend("hdfs:///fink-checkpoints", true));  

注意：除了HDFS之外，如果在FLINK_HOME/plugins下添加了相應(yīng)的依賴項，那么還可以使用其他本地或基于云的對象存儲。

最佳實踐和高級配置

我們希望這個概述能幫助您更好地理解RocksDB在Flink中的作用，以及如何使用RocksDBStateBackend成功地運行作業(yè)。最后，我們將探討一些最佳實踐和一些參考點，以便進一步進行故障診斷和性能調(diào)優(yōu)。

狀態(tài)在RocksDB中的位置

如前所述，RocksDBStateBackend 中的運行中狀態(tài)會溢出到磁盤上的文件。這些文件位于Flink配置指定的state.backend.rocksdb.localdir目錄下。因為磁盤性能直接影響RocksDB的性能，所以建議將此目錄放在本地磁盤上。不鼓勵將其配置到基于網(wǎng)絡(luò)的遠程位置，如NFS或HDFS，因為寫入遠程磁盤通常比較慢。高可用性也不是飛行狀態(tài)(in-flight state)的要求。如果需要高磁盤吞吐量，則首選本地SSD磁盤。

狀態(tài)快照將持久化到遠程持久存儲。在狀態(tài)快照期間，TaskManager會對飛行中的狀態(tài)(in-flight state)進行快照并遠程存儲。將狀態(tài)快照傳輸?shù)竭h程存儲完全由TaskManager本身處理，而不需要狀態(tài)后端的參與。所以，state.checkpoints.dir 目錄或者您在代碼中為特定作業(yè)設(shè)置的參數(shù)可以是不同的位置，如本地HDFS集群或基于云的對象存儲，如Amazon S3、Azure Blob Storage、Google cloud Storage、Alibaba OSS等。

RocksDB故障診斷

要檢查RocksDB在生產(chǎn)中的行為，應(yīng)該查找名為LOG的RocksDB日志文件。默認(rèn)情況下，此日志文件與數(shù)據(jù)文件位于同一目錄中，即Flink配置指定的目錄state.backend.rocksdb.localdir。啟用時，RocksDB統(tǒng)計信息也會記錄在那里，以幫助診斷潛在的問題。有關(guān)更多信息，請查看RocksDB Wiki中的Troubleshooting Guide。如果你對RocksDB行為趨勢感興趣，可以考慮為你的Flink作業(yè)啟用RocksDB本機指標(biāo)。

注意：從Flink1.10開始，通過將日志級別設(shè)置為HEADER，RocksDB日志記錄被有效地禁用。要啟用它，請查看How to get RocksDB’s LOG file back for advanced troubleshooting。

警告：在Flink中啟用RocksDB的原生指標(biāo)可能會對您的工作產(chǎn)生負(fù)面影響。

從Flink 1.10開始，F(xiàn)link默認(rèn)將RocksDB的內(nèi)存分配配置為每個任務(wù)槽的托管內(nèi)存(managed memory)量。改善內(nèi)存相關(guān)性能問題的主要機制是通過Flink配置taskmanager.memory.managed.size或taskmanager.memory.managed.fraction增加Flink的托管內(nèi)存。對于更細(xì)粒度的控制，應(yīng)該首先通過設(shè)置state.backend.rocksdb.memory.managed為false，然后從以下Flink配置開始：state.backend.rocksdb.block.cache-size（與RocksDB中的塊大小相對應(yīng)），state.backend.rocksdb.writebuffer.size（與RocksDB中的write_buffer_size相對應(yīng)），以及state.backend.rocksdb.writebuffer.count（對應(yīng)于RocksDB中的最大寫入緩沖區(qū)數(shù)）。有關(guān)更多詳細(xì)信息，請查看這篇關(guān)于如何在Flink中管理RocksDB內(nèi)存大小的文章和RocksDB內(nèi)存使用Wiki頁面。

在RocksDB中寫入或覆蓋數(shù)據(jù)時，RocksDB線程在后臺管理從內(nèi)存到本地磁盤的刷新和數(shù)據(jù)壓縮。在多核CPU的機器上，應(yīng)該通過設(shè)置Flink配置state.backend.rocksdb.thread.num（對應(yīng)于RocksDB中的max_background_jobs）來增加后臺刷新和壓縮的并行性。對于生產(chǎn)設(shè)置來說，默認(rèn)配置通常太小。如果你的工作經(jīng)常從RocksDB讀取內(nèi)容，那么應(yīng)該考慮啟用布隆過濾器。

對于其他RocksDBStateBackend配置，請查看Flink文檔Advanced RocksDB State Backends Options。有關(guān)進一步的調(diào)優(yōu)，請查看RocksDB Wiki中的RocksDB Tuning Guide。

RocksDB狀態(tài)后端（即RocksDBStateBackend）是Flink中捆綁的三種狀態(tài)后端之一，在配置流應(yīng)用程序時是一個很好的選擇。它使可擴展的應(yīng)用程序能夠保持高達數(shù)TB的狀態(tài)，并保證exactly-once。如果Flink作業(yè)的狀態(tài)太大，無法放入JVM堆中，或者你對增量檢查點很感興趣，或者希望有可預(yù)測的延遲，那么應(yīng)該使用RocksDBStateBackend。由于RocksDB作為本機線程嵌入到TaskManager進程中，并且可以處理本地磁盤上的文件，RocksDBStateBackend支持開箱即用，無需更多設(shè)置和管理任何外部系統(tǒng)或進程。

關(guān)于如何在Apache Flink中使用RocksDB狀態(tài)后端問題的解答就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道了解更多相關(guān)知識。