Flink 原理与实现：State

逃离我推掉我的手 2023-06-16 14:57 22阅读 0赞

扫码关注公众号免费阅读全文：**冰山烈焰的黑板报**  
![在这里插入图片描述][20200906232656299.jpg_pic_center]

# 1. State #

State 是流计算中非常重要的一个概念。首先，我们区分下有状态的流计算和无状态的流计算：

*  **Stateless Stream Processing（无状态的流计算）**：算子仅考虑当前的输入，而无需进一步了解上下文或历史的计算信息。
 *  **Stateful Stream Processing（有状态的流计算）**：算子存储的过去的计算信息，会影响将来的输入处理。

举个例子，现在我们假设有个数据结构如下：

case class WordCount(word: String, frequency: Long = -999L)

有以下两个需求：

1.  过滤掉出现频次小于 1 的 word。
2.  统计每个 word 出现的频次。

对于【需求 1】，只用比较当前 word 的 frequency 是否大于 0，而不用考虑历史的计算信息。然后将 frequency 是否大于 0 的 word 发送到下游 Sink 即可。这便是 **Stateless Stream Processing**。

对于【需求2】，算子需要用当前 word 的 frequency 加上算子中存储的历史计算过的该 word 出现的 frequency。因此，这是 **Stateful Stream Processing**。

其实，State 在数据处理中是比较常见的。比如，统计每天最高温度时，需要存储当天历史的最高温度；计算 PV、UV、DAU 等等都需要 State。

在流处理任务中，在事件之间持久化的 State，我们甚至可以将其视为编程模型中的一等公民。鉴于此，我们需要额外的系统来管理流的 State，即使会导致些许延迟。

另外，由于流计算通常会处理无界数据，会使 State 无限制地增长。因此，为了限制 State 的大小，一般 State 只维护事件的核心概要信息，例如 count、sum 和一些简单的数据结构。

如你所见，实现 State 会遇到以下一些挑战：

1.  State 的管理。系统需要高效管理 State，并且保证在并发操作中的正确性。
2.  State 的分配。在并行计算中，State 可以根据 Key 划分或者保证每个 Partition 的 State 彼此相互独立。
3.  State 的恢复。State 需要能够正确恢复，并且保证恢复后的计算结果仍然正确。

# 2. State Management #

业界很多实时计算引擎都包含了 **Stateful Stream Processing** 特征，比如 Google Dataflow、Spark（Structure）Streaming、Kafka Streams 都对提供了对内置 State 的支持，Flink 也不例外。这一节主要讨论 Flink 支持的 State 类型，State 是怎么存储和维护的，以及怎么在分布式系统中扩展的。

大体上，任务维护的所有用于计算函数结果的数据均属于任务状态。你甚至可以将 State 想象成任务逻辑访问的局部变量。图1 展示了任务和 State 之间的交互过程。  
![图1 A stateful stream processing task][1 A stateful stream processing task]  
在上图中，一个任务接收输入数据，当处理输入数据时，会读取 State，并且根据输入数据和 State 计算结果，同时会更新 State。这个例子中对 State 的读写都很直观。但是，高效可靠的管理 State 却是充满挑战的。这包括操作大量数据、内存计算、失败时 State 不丢失、State 的一致性，这些都是 Flink 需要考虑的，从而可以让开发人员只需要聚焦于业务逻辑的处理上。

在 Flink 中，State 是一种特殊的算子。在使用 State 之前，算子需要注册 State。目前，Flink 包括两种 State：

1.  Operator State
2.  Keyed State

## 2.1 Operator State ##

**Operator State** 的作用范围仅限于并行的算子。这意味着同一并行任务处理的所有记录都可以访问同一 State。**Operator State** 无法被相同或不同算子的其他任务访问。图2 展示了任务是如何访问 Operator State 的。  
![图2 Tasks with operator state][2 Tasks with operator state]  
Operator State 有以下 3 种类型：

1.  List State
2.  Union List State
3.  Broadcast State

## 2.2 Keyed State ##

**Keyed State** 维护和访问根据算子的输入流中定义的键对应的 State。Flink 为每个键维护一个 State 实例，并将具有相同键的所有数据划分到维护该键 State 的算子中。因此，相同键的所有数据可以访问同一 State。图3 展示任务是如何和 **Keyed State** 交互的。  
![图3 Tasks with keyed state][3 Tasks with keyed state]  
你可以将 Keyed State 看作是一个 Key-Value 的 Map。Flink 提供了一下内置的 Keyed State：

1.  Value State
2.  List State
3.  Map State
4.  Reducing State
5.  Aggregating State

## 2.3 State Backends ##

由于 **Stateful Stream Processing** 需要为每一条数据读取和更新 State，因此，低延迟的访问 State 是至关重要的。如何存储、访问和维护 State 的这个组件就是 **State Backend**。**State Backend** 的职责有两个：

1.  管理本地 State；
2.  Checkpoint State 到远程。

关于管理本地 State，Flink 提供了两种 **State Backend**：

1.  内存存储。这种 **State Backend** 通过把 State 对象放到 JVM Heap 的内存中，可以实现非常快的 State 访问，但受限于内存的大小；
2.  RocksDB 存储。这种 **State Backend** 会把 State 对象序列化之后，存到 RocksDB中，由 RocksDB 写入到本地磁盘。这种方案虽然会使 State 访问更慢些，但是却可以支持更大的 State。

State Checkpointing 对于 Flink 这样的分布式系统而言，是非常重要的，因为 State 仅在本地维护。由于 TaskManager 可能由于各种原因失败，所以，State 的存储必须考虑易失性。如此，远程存储必须是一个分布式文件系统或者数据库系统。但是，不同的 **State Backend** 在 Checkpoint 的方式上会有所不同。例如，采用 RocksDB State Backend 可以支持增量 Checkpoint，这样可以减小 State 大小的瓶颈（这部分笔者会用专门的文章讨论）。

## 2.4 Scaling Stateful Operators ##

Flink 作为一款分布式系统，无疑面对横向扩展，算子也会面临并行度的增减。调整 Stateful 算子的并行度是一个很大的挑战，因为这些算子的 State 需要 Repartition，以分配更多或更少的并行任务。Fink 对于不同的 State 类型，采用的 4 种不同的策略。

#### Scaling Keyed State ####

**Keyed State 算子（operators with keyed state）** 扩展的时候不是针对每个单独的 Key 进行 Repartition。但是，为了提高效率，Flink 将这些 Key 组织成 key-group。一个 key-group 是这些 Key 的一个分区，而 Flink 也是按照 key-group 去 Repartition 的。图 4 展示了 **Keyed State 算子** 的扩展过程。  
![图4 scaling an operator with keyed state out and in][4 scaling an operator with keyed state out and in]

#### Scaling Operator List State ####

**Operator List State 算子（operators with operator list state）** 是将 List 中的 State 实体都重新分配。换言之，所有并行算子的 List 中的 State 实体被收集起来，然后重新分布到更多或更少的任务中去。如果 List 中的 State 实体数目少于新的并行算子，那么有些任务将会填充空的 State。图5 展示了 **Operator List State 算子** 的重分布过程。  
![图5 scaling an operator with operator list state out and in][5 scaling an operator with operator list state out and in]

#### Scaling Operator Union List State ####

**Operator Union List State 算子（operator with operator union list state）** 扩展采用广播的方式，它将所有 List 的 State 实体广播给每个任务，让任务自己去裁决使用或废弃哪些 State。图6 表示 **Operator Union List State 算子** 的 Repartition 过程。  
![图6 scaling an operator with operator union list state out and in][6 scaling an operator with operator union list state out and in]

#### Scaling Operator Broadcast State ####

**Operator Broadcast State 算子（operator with operator broadcast state）** 采用复制给所有新的任务复制 State 的策略实现扩展。这样做的原因是广播 State 可以确保每个任务都有相同的 State。防止任务扩展过程中 State 丢失。**Operator Broadcast State 算子** 的扩展过程可以参考图7。  
![图7 scaling an operator with operator broadcast state out and in][7 scaling an operator with operator broadcast state out and in]

# 3. 总结 #

Stream Processing 分为 **Stateless Stream Processing** 和 **Stateful Stream Processing** 两种。**Stateful Stream Processing** 是当前业界流程的流计算引擎的重要特征之一，Flink 也不例外。Flink 支持 **Operator State** 和 **Keyed State** 两大类 State，而且各自又分为若干种 State。Flink 使用 **State Backend** 对 State 进行管理——存储、访问和维护，同时 **State Backend** 有多种，需要根据具体需求权衡选择。Scaling State 具有很大的挑战，Flink 采用了 4 种不同的策略，分别对不同类型的 State 进行 Scaling。

[20200906232656299.jpg_pic_center]: https://img-blog.csdnimg.cn/20200906232656299.jpg#pic_center
[1 A stateful stream processing task]: https://img-blog.csdnimg.cn/20191127232033861.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[2 Tasks with operator state]: https://img-blog.csdnimg.cn/20191128000015183.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[3 Tasks with keyed state]: https://img-blog.csdnimg.cn/20191128112953479.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[4 scaling an operator with keyed state out and in]: https://img-blog.csdnimg.cn/20191129214850231.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[5 scaling an operator with operator list state out and in]: https://img-blog.csdnimg.cn/20191129215917207.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[6 scaling an operator with operator union list state out and in]: https://img-blog.csdnimg.cn/20191129220610797.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[7 scaling an operator with operator broadcast state out and in]: https://img-blog.csdnimg.cn/20191129221237974.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9nYXRzYnluZXd0b24uYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70