1.flink 常用时间窗口示意图

2.flink checkpoint状态储存三种方式选择

Flink 提供了三种可用的状态后端：MemoryStateBackend，FsStateBackend，和RocksDBStateBackend。

(1)MemoryStateBackend

MemoryStateBackend 是将状态维护在 Java 堆上的一个内部状态后端。键值状态和窗口算子使用哈希表来存储数据（values）和定时器（timers）。当应用程序 checkpoint 时，此后端会在将状态发给 JobManager 之前快照下状态，JobManager 也将状态存储在 Java 堆上。默认情况下，MemoryStateBackend 配置成支持异步快照。异步快照可以避免阻塞数据流的处理，从而避免反压的发生。

使用 MemoryStateBackend 时的注意点：

默认情况下，每一个状态的大小限制为 5 MB。可以通过 MemoryStateBackend 的构造函数增加这个大小。

状态大小受到 akka 帧大小的限制，所以无论怎么调整状态大小配置，都不能大于 akka 的帧大小。也可以通过 akka.framesize 调整 akka 帧大小（通过配置文档了解更多）。

状态的总大小不能超过 JobManager 的内存。

何时使用 MemoryStateBackend：

本地开发或调试时建议使用 MemoryStateBackend，因为这种场景的状态大小的是有限的。

MemoryStateBackend 最适合小状态的应用场景。例如 Kafka consumer，或者一次仅一记录的函数 （Map, FlatMap，或 Filter）。

(2) FsStateBackend

FsStateBackend 需要配置的主要是文件系统，如 URL（类型，地址，路径）。举个例子，比如可以是：“hdfs://namenode:40010/flink/checkpoints” 或“s3://flink/checkpoints”

当选择使用 FsStateBackend 时，正在进行的数据会被存在 TaskManager 的内存中。在 checkpoint 时，此后端会将状态快照写入配置的文件系统和目录的文件中，同时会在 JobManager 的内存中（在高可用场景下会存在 Zookeeper 中）存储极少的元数据。

默认情况下，FsStateBackend 配置成提供异步快照，以避免在状态 checkpoint 时阻塞数据流的处理。该特性可以实例化 FsStateBackend 时传入 false 的布尔标志来禁用掉，例如：

new FsStateBackend(path, false);

使用 FsStateBackend 时的注意点：

当前的状态仍然会先存在 TaskManager 中，所以状态的大小不能超过 TaskManager 的内存。

何时使用 FsStateBackend：

FsStateBackend 适用于处理大状态，长窗口，或大键值状态的有状态处理任务。

FsStateBackend 非常适合用于高可用方案。

(3)RocksDBStateBackend

RocksDBStateBackend 的配置也需要一个文件系统（类型，地址，路径），如下所示：

“hdfs://namenode:40010/flink/checkpoints” 或“s3://flink/checkpoints”

RocksDB 是一种嵌入式的本地数据库。RocksDBStateBackend 将处理中的数据使用 RocksDB 存储在本地磁盘上。在 checkpoint 时，整个 RocksDB 数据库会被存储到配置的文件系统中，或者在超大状态作业时可以将增量的数据存储到配置的文件系统中。同时 Flink 会将极少的元数据存储在 JobManager 的内存中，或者在 Zookeeper 中（对于高可用的情况）。RocksDB 默认也是配置成异步快照的模式。

使用 RocksDBStateBackend 时的注意点：

RocksDB 支持的单 key 和单 value 的大小最大为每个 2^31 字节。这是因为 RocksDB 的 JNI API 是基于 byte[] 的。

我们需要强调的是，对于使用具有合并操作的状态的应用程序，例如 ListState，随着时间可能会累积到超过 2^31 字节大小，这将会导致在接下来的查询中失败。

何时使用 RocksDBStateBackend：

RocksDBStateBackend 最适合用于处理大状态，长窗口，或大键值状态的有状态处理任务。

RocksDBStateBackend 非常适合用于高可用方案。

RocksDBStateBackend 是目前唯一支持增量 checkpoint 的后端。增量 checkpoint 非常使用于超大状态的场景。

当使用 RocksDB 时，状态大小只受限于磁盘可用空间的大小。这也使得 RocksDBStateBackend 成为管理超大状态的最佳选择。使用 RocksDB 的权衡点在于所有的状态相关的操作都需要序列化（或反序列化）才能跨越 JNI 边界。与上面提到的堆上后端相比，这可能会影响应用程序的吞吐量。