Flink 状态和容错

状态概念

状态

状态是什么：

其实状态并不是flink独有的特性，状态在生活中普遍存在和使用，只是近些年在实时任务这些词汇会被频繁提及

简单举几个程序开发中会使用状态的例子：

• java用set集合去重
• 离线spark rdd的cache和checkpoint
• 离线spark Remote Shuffle Service
• Flink 实时计算dau，需要对uid去重，就要记录历史所有来过的uid

离线其实也有状态、状态管理，比如Remote Shuffle Service，对于整个任务来说，每个 Stage 的结果就是状态，而Remote Shuffle Service 就起到了*** "管理" 状态 ***的作用

状态解决了什么：

1. 离线任务失败重启：失败重新读一遍输入数据，然后重新计算一遍即可
2. 实时任务失败重启：实时任务一般都是 7x24 小时 long run 的，挂了之后，就会有以下两个问题。

场景示例：一个消费上游 Kafka，使用 Set去重计算 DAU 的实时任务。

• 数据质量问题：当这个实时任务挂了之后恢复，Set空了，这时候任务再继续从上次失败的 Offset 消费 Kafka 产出数据，则产出的数据就是错误的
• 数据时效问题：如果从头消费数据，可能需要回溯几个小时，有较长一段时间看到的结果是错误的，而且如果超出kafka保存时间，还会有第一个问题

状态管理

状态管理又是什么：

只有state没有用，需要把 state 管理起来，在任务 failover 后将状态恢复（比如说 Remote Shuffle Service 把数据管理到远程，失败时可以远程恢复），保障数据质量、时效

协调整个任务 when，how 去将 Flink 任务本地机器中存储在状态后端的状态去同步到远程文件存储系统（比如 HDFS）的过程就是Checkpoint 或 Savepoint

状态后端

Flink 提供了 3 种状态后端用于管理和存储状态数据

1. MemoryStateBackend
2. FsStateBackend
3. RocksdbStateBackend

MemoryStateBackend

运行时所需的 State 数据全部保存在 TaskManager JVM 堆上内存中，执行 Checkpoint 的时候，会把 State 的快照数据保存到 JobManager 进程 的内存中。执行 Savepoint 时，可以把 State 存储到文件系统中

效率好，但是不安全，生产一般不用

FSStateBackend

原理：运行时所需的 State 数据全部保存在 TaskManager 的内存中，执行 Checkpoint 的时候，会把 State 的快照数据保存到配置的文件系统中。TM 是异步将 State 数据写入外部存储。

优点：效率比RocksDBStateBackend要好，但是State 大小不能超过 TM 内存（会OOM），适合小状态、短窗口、或者小键值状态任务（比如明细层 ETL 任务，小时间间隔的 TUMBLE 窗口）

RocksDBStateBackend

原理：使用嵌入式的本地数据库 RocksDB 将流计算数据状态存储在本地磁盘中。在执行 Checkpoint 的时候，会将整个 RocksDB 中保存的 State 数据全量或者增量持久化到配置的文件系统中。

优点：支持增量检查点、适合大状态

Rocksdb 用磁盘存储 State，所以会涉及到访问 State 磁盘序列化、反序列化，相同任务使用 Filesystem 性能为 Rocksdb 的 n 倍

容错

当我们启停任务时，会自动SavePoint，把状态保存到文件，下次启动可以恢复

如果任务失败，就需要从 CheckPoint 恢复数据，CheckPoint 是状态的快照，设置 CheckPoint 5 min，每 5 min 做一次 CheckPoint，把状态放到远程文件（MemoryStateBackend 是内存，所以会状态会丢失）

如果任务失败，就可以从上次的 CheckPoint 恢复

状态分类

Flink 中的状态分类有两大类，Managed State 和 Raw State（Raw State基本用不到）

Managed State 细分，它又有两种类型：operator-state 和 keyed-state