Flink状态流处理：State Backends三种方式详解

本帖最后由 pig2 于 2018-10-29 15:49 编辑
问题导读
1.什么是State Backends？
2.State Backends有什么作用？
3.State Backends分为哪些类？
4.State Backends分别都是什么情况下使用?
5.State Backends各自的优缺点是什么？

最新经典文章，欢迎关注公众号

Backends知识点补充：
当检查点（checkpoint）机制启动时，状态将在检查点中持久化来应对数据丢失以及恢复。而状态在内部是如何表示的、状态是如何持久化到检查点中以及持久化到哪里都取决于选定的State Backend。

可用的State Backends
Flink自带了三种state backend：

·  MemoryStateBackend

·  FsStateBackend

·  RocksDBStateBackend

在没有配置的情况下，系统默认使用MemoryStateBackend

状态解释：
使用Data Stream API编写的程序通常以多种形式维护状态：

·  窗口将收集element或在它被触发后聚合element

·  Transformation方法可能会使用key/value状态接口来存储值

·  Transformation方法也可能会实现Checkpointed接口来使其本地变量进入容错机制

这篇文章探讨Flink状态流处理，更确切地说是Flink中可用的不同Backends。在以下部分中，我们将介绍Flink的3个Backends，它们的局限性以及何时根据特定于案例的要求使用它们。
通过有状态流处理，当开发人员启用Flink应用程序的检查点时，状态将持续存在以防止数据丢失并确保在发生故障时完全恢复。为应用程序选择Backends将影响状态持久化的方式和位置。

上面我们提到Flink附带三个可用的StateBackend：MemoryStateBackend，FsStateBackend和RocksDBStateBackend。

MemoryStateBackend
MemoryStateBackend是一个内部状态backend ，用于维护Java堆上的状态。 Key/value 状态和窗口运算符包含存储值和计时器的哈希表。
当应用程序检查点时，此backend 将状态发送到Flink的作业管理器之前拍摄状态的快照，该作业管理器也将其存储在Java堆上。
默认情况下，MemoryStateBackend配置为支持异步快照。异步快照可避免可能导致流应用程序背压的潜在阻塞管道。

使用MemoryStateBackend时需要注意什么：

默认情况下，每个个体的状态的大小限制为5 MB。可以在MemoryStateBackend构造函数中进一步增加大小。
状态大小受akka帧大小的限制，无论在配置中设置为最大状态size ，都不能大于akka帧大小（可以在配置中找到更多信息）。
聚合状态必须适合JobManager内存。

何时使用MemoryStateBackend：

建议使用MemoryStateBackend进行本地开发或调试，因为它的状态有限
MemoryStateBackend最适合具有小状态大小的用例和有状态流处理应用程序，例如仅包含一次记录功能（Map，FlatMap或Filter）的作业或使用Kafkaconsumer.。

FsStateBackend
FsStateBackend配置使用文件系统完成，例如URL（类型，地址，路径）。一些示例文件系统可能是：

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

当选择FsStateBackend时，正在传输的数据保存在任务管理器（ Task Manager）的内存中。在检查点上，此backend 将状态快照写入系统配置的文件和目录中的文件，同时它将在JobManager的内存或Zookeeper中存储最少的元数据（对于高可用性情况）。

默认情况下，FsStateBackend配置为提供异步快照，以避免在写入状态检查点时阻塞处理管道（processing pipeline）。可以通过将构造函数中相应的boolean标志设置为false来禁用该功能，例如：

new FsStateBackend(path, false);
复制代码

何时使用FsStateBackend：

FsStateBackend最适合处理大状态，长窗口或大键/值状态的Flink有状态流处理作业。
FsStateBackend最适合每个高可用性设置。

RocksDBStateBackend
使用文件系统（类型，地址，路径）执行RocksDBStateBackend的配置，如下例所示：

“hdfs://namenode:40010/flink/checkpoints” or
“s3://flink/checkpoints”.

RocksDBStateBackend使用RocksDB数据库在本地磁盘上保存传输中的数据。在检查点上，整个RocksDB数据库将被检查点到（ checkpointed into）配置的文件系统中，或者在非常大的状态作业的情况下增量差异。同时，Apache Flink将一些最小的元数据存储在JobManager的内存或Zookeeper中（对于高可用性情况）。 RocksDB默认配置为执行异步快照。

使用RocksDBStateBackend时需要注意什么：

RocksDB的每个密钥和每个值的最大支持大小为每个2 ^ 31个字节。这是因为RocksDB的JNI桥API基于byte []。
我们需要在此强调，对于使用具有合并操作的状态（例如ListState）的有状态流处理应用程序，可以累积超过2 ^ 31字节超时的值size，这将导致它们在任何后续检索时失败。

何时使用RocksDBStateBackend：

RocksDBStateBackend最适合处理大状态，长窗口或大键/值状态的Flink有状态流处理作业。
RocksDBStateBackend最适合每个高可用性设置。
RocksDBStateBackend是目前唯一可用于支持有状态流处理应用程序的增量检查点的状态后端。

使用RocksDB时，状态大小仅受可用磁盘空间量的限制，这使RocksDBStateBackend成为管理超大状态的绝佳选择。使用RocksDB时的权衡是所有状态访问和检索都需要序列化（或反序列化）才能跨越JNI边界。与上面提到的 on-heap（内存） backends相比，这可能会影响应用程序的吞吐量。
不同的状态backends 服务于多个开发人员要求，应在开始开发应用程序之前仔细考虑和进行广泛规划后选择。这可确保选择正确的状态backends 以最好地满足应用程序和业务需求。