title: Flink系列

一、Flink StateBackend 深入剖析和應(yīng)用

? StateBackend 定義了狀態(tài)是如何存儲(chǔ)的，不同的 State Backend 會(huì)采用不同的方式來存儲(chǔ)狀態(tài)，核心入口是： StateBackend， Flink 提供了三種不同形式的存儲(chǔ)后端，分別是 MemoryStateBackend, FsStateBackend 和 RocksDBStateBackend。

• MemoryStateBackend 會(huì)將工作狀態(tài)（Task State）存儲(chǔ)在 TaskManager 的內(nèi)存中，將檢查點(diǎn)（Job State）存儲(chǔ)在 JobManager 的內(nèi)存中，速度很快，不支持持久化，通常用來存儲(chǔ)一些 state 量小的情況下的 state。這種方式是非常不安全的，且受限于JobManager的內(nèi)存大小，主要在開發(fā)調(diào)試中使用。

• FsStateBackend 會(huì)將工作狀態(tài)存儲(chǔ)在 TaskManager 的內(nèi)存中，將檢查點(diǎn)存儲(chǔ)在文件系統(tǒng)中（通常是分布式文件系統(tǒng)），用來存儲(chǔ) state 量比較大的，window 窗口很長的一些 job 的 state 比較合適。生產(chǎn)環(huán)境常用此方案。

• RocksDBStateBackend 會(huì)把狀態(tài)存儲(chǔ)在 RocksDB 中，將檢查點(diǎn)存儲(chǔ)在文件系統(tǒng)中（類似 FsStateBackend），和 MemoryStateBackend 對(duì)比是速度快，GC 少，支持異步 Snapshot 持久化。用來存儲(chǔ) state 量比較大的，window 窗口很長的一些 job 的 state 比較合適。

? 綜上所述，MemoryStateBackend 和 FsStateBackend 都是在內(nèi)存中進(jìn)行狀態(tài)管理，所以可以獲取較低的讀寫延遲，但會(huì)受限于 TaskManager 的內(nèi)存大??；而RocksDBStateBackend 直接將 State 存儲(chǔ)到 RocksDB 數(shù)據(jù)庫中，所以不受 JobManager 的內(nèi)存限制，但會(huì)有讀寫延遲，同時(shí) RocksDBStateBackend 支持增量備份，這是其他兩個(gè)都不支持的特性。一般來說，如果不是對(duì)延遲有極高的要求，RocksDBStateBackend 是更好的選擇。

? 淘汰掉原來的三種實(shí)現(xiàn)，提供兩種新的實(shí)現(xiàn)的目的：為了接口統(tǒng)一！底層原理沒變。window編程也被統(tǒng)一了，Time編程也被統(tǒng)一了。

配置：
state.backend: hashmap
state.checkpoint-storage: jobmanager
state.checkpoints.dir: hdfs://hadoop10/flink/checkpoints
state.savepoints.dir: hdfs://hadoop10/flink/savepoints

細(xì)粒度：Task State： 一個(gè) Application 會(huì)運(yùn)行很多的 Task, 每個(gè) Task 運(yùn)行的時(shí)候，都有自己的狀態(tài)， 故障轉(zhuǎn)移 = FailOverStrategy

• 要么是 TaskManager 的堆內(nèi)存

• 要么是 RocksDB 中

粗粒度：Job State：在某個(gè)時(shí)候，通過某種手段（checkpoint）把這個(gè) job 的所有 Task 的 state 做一個(gè)持久化，就形成了 job 的 state， 重啟策略 = RestartStrategy

• 要么是 JobManager 的堆內(nèi)存

• 要么是外部的高可用系統(tǒng)中，可以是HDFS

Flink StateBackend 的三種實(shí)現(xiàn)對(duì)比：

1.1 MemoryStateBackend

? 默認(rèn)情況下，狀態(tài)信息是存儲(chǔ)在 TaskManager 的堆內(nèi)存中的，checkpoint 的時(shí)候?qū)顟B(tài)保存到 JobManager 的堆內(nèi)存中。

缺點(diǎn)：
	只能保存數(shù)據(jù)量小的狀態(tài)
	狀態(tài)數(shù)據(jù)有可能會(huì)丟失
優(yōu)點(diǎn)：
	開發(fā)測試很方便

1.2 FSStateBackend

狀態(tài)信息存儲(chǔ)在 TaskManager 的堆內(nèi)存中的，checkpoint 的時(shí)候?qū)顟B(tài)保存到指定的文件中 (HDFS 等文件系統(tǒng))

缺點(diǎn)：
	狀態(tài)大小受TaskManager內(nèi)存限制(默認(rèn)支持5M)
優(yōu)點(diǎn)：
	狀態(tài)訪問速度很快
	狀態(tài)信息不會(huì)丟失
用于：
	生產(chǎn)，也可存儲(chǔ)狀態(tài)數(shù)據(jù)量大的情況

1.3 RocksDBStateBackend

? 狀態(tài)信息存儲(chǔ)在 RocksDB 數(shù)據(jù)庫 (key-value 的數(shù)據(jù)存儲(chǔ)服務(wù))， 最終保存在本地文件中。checkpoint 的時(shí)候?qū)顟B(tài)保存到指定的文件中 (HDFS 等文件系統(tǒng))

缺點(diǎn)：
	狀態(tài)訪問速度有所下降
優(yōu)點(diǎn)：
	可以存儲(chǔ)超大量的狀態(tài)信息
	狀態(tài)信息不會(huì)丟失
用于：
	生產(chǎn)，可以存儲(chǔ)超大量的狀態(tài)信息

二、Flink StateBackend 原理剖析與實(shí)踐

第一種：單任務(wù)調(diào)整

修改當(dāng)前任務(wù)代碼
env.setStateBackend(new FsStateBackend("hdfs://hadoop10/flink/checkpoints"));
env.setStateBackend(new MemoryStateBackend());
env.setStateBackend(new RocksDBStateBackend(filebackend, true));

第二種：全局調(diào)整

修改 flink-conf.yaml
state.backend: filesystem
state.checkpoints.dir: hdfs://hadoop10/flink/checkpoints

注意：state.backend的值可以是下面幾種：
1、jobmanager(MemoryStateBackend)
2、filesystem(FsStateBackend)
3、rocksdb(RocksDBStateBackend)

MemoryStateBackend（老版本的默認(rèn)實(shí)現(xiàn)） 和 FsStateBackend 的代碼寫法，其實(shí)它們已經(jīng)被廢棄，建議使用：HashMapStateBackend（新版本的默認(rèn)實(shí)現(xiàn)）

用的是HashMapStateBackend，但是給job級(jí)別的數(shù)據(jù)保存到 Job Manager 的堆內(nèi)內(nèi)存中：

// HashMapStateBackend 替代 MemoryStateBackend
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 1、設(shè)置使用 HashMapStateBackend，Task State 存儲(chǔ)在 TaskManager 的堆內(nèi)存中
env.setStateBackend(new HashMapStateBackend());
// 2、這樣設(shè)置 checkpoint 的 state 存儲(chǔ)方式：把 job State 存儲(chǔ)在 JobManager 的堆內(nèi)存中
env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());

用的是HashMapStateBackend，但是給job級(jí)別的數(shù)據(jù)保存到 Job Manager 的外面的高可用系統(tǒng)HDFS中：

// HashMapStateBackend 替代 FsStateBackend
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 1、設(shè)置使用 HashMapStateBackend，Task State 存儲(chǔ)于 TaskManager 堆內(nèi)存中
env.setStateBackend(new HashMapStateBackend());
// 2、需要設(shè)置外部高可用文件系統(tǒng)存儲(chǔ)路徑用來保存 Job State
env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop10/flink/checkpoints");

RocksDBStateBackend 代碼寫法，其實(shí) RocksDBStateBackend 也已經(jīng)被廢棄，建議使用 EmbeddedRocksDBStateBackend

// EmbeddedRocksDBStateBackend
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 1、設(shè)置 EmbeddedRocksDBStateBackend，Task State 存儲(chǔ)在 RocksDB 中（內(nèi)存+磁盤）
env.setStateBackend(new EmbeddedRocksDBStateBackend());
// 2、設(shè)置外部高可用文件系統(tǒng)存儲(chǔ)路徑用來保存 Job State
env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop10/flink/checkpoints");

如果使用 RocksDB 的方式，需要引入依賴：

<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-statebackend-rocksdb -->
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-statebackend-rocksdb_2.12</artifactId>
	<version>1.14.3</version>
	<scope>test</scope>
</dependency>

聲明：
????????文章中代碼及相關(guān)語句為自己根據(jù)相應(yīng)理解編寫，文章中出現(xiàn)的相關(guān)圖片為自己實(shí)踐中的截圖和相關(guān)技術(shù)對(duì)應(yīng)的圖片，若有相關(guān)異議，請(qǐng)聯(lián)系刪除。感謝。轉(zhuǎn)載請(qǐng)注明出處，感謝。

B站： https://space.bilibili.com/1523287361 點(diǎn)擊打開鏈接
微博地址： http://weibo.com/luoyepiaoxue2014 點(diǎn)擊打開鏈接文章來源地址http://www.zghlxwxcb.cn/news/detail-577284.html

到了這里，關(guān)于Flink系列之Flink中StateBackend深入剖析和應(yīng)用的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！