在流计算场景中，数据会不断流入Apache Flink系统，每一条进入Apache Flink系统的数据都会触发计算。如果要做Count聚合计算，是每次触发计算都重新计算历史上所有的流入数据，还是每次都根据上一次的计算结果做增量计算？答案是肯定的，Apache Flink根据上一次计算结果进行增量计算。那么问题来了:“最后的计算结果保存在哪里，能保存在内存中吗？”答案是否定的。如果存储在内存中，当一个计算节点由于网络、硬件等原因出现故障时，最后的计算结果就会丢失。当节点恢复时，需要重新计算历史上所有的数据(可能是十天或几百天的数据)，所以为了避免这个灾难性的问题，Apache Flink会使用State来存储计算结果。本文将向您介绍阿帕奇弗林克州。

什么是国家？

这个问题好像有点“弱智”？不管问题的答案是否明显，我还是想简单说一下弗林克是什么州。状态是指流计算过程中计算节点的中间计算结果或元数据属性。例如，在汇总过程中，中间汇总结果应记录在State中。例如，当使用Apache Kafka作为数据源时，我们还应该记录读取记录的偏移量，这些状态数据将在计算过程中被持久化(插入或更新)。因此，Flink中的状态是Flink任务内部数据(计算数据和元数据属性)的快照，与时间相关。

你为什么需要国家？

与批处理计算相比，状态是流计算独有的。批处理计算没有故障转移机制，因此它要么成功，要么重新计算。在大多数情况下，计算是递增的，数据是逐个处理的(大多数情况)。每次计算都是根据最后的计算结果进行的。该机制绑定存储最后的计算结果(生产模式应该是持久的)。此外，由于机器、网络、脏数据等原因导致的程序错误，在重新启动作业时，需要从成功的检查点(后面章节会介绍)恢复状态。增量计算和故障转移都需要状态的支持。

状态存储实现

Flink中有三种状态的存储实现，如下所示:

基于内存的heapstate后端-在调试模式下使用，不建议在生产模式下使用；

fsstatefender——基于HDFS的分布式文件持久化，每次读写都操作内存，也需要考虑OOM

基于RocksDB的RocksDB后端——本地文件+异步HDFS持久化；

状态存储模式

默认情况下，Apache Flink将State存储在RocksDB+HDFS。状态存储分为两个阶段:首先，它被本地存储在RocksDB中，然后异步同步到远程HDFS。这种设计不仅消除了heapstate后端的限制(内存大小、机器损坏和丢失等)。)，还降低了纯分布式存储的网络IO开销。

状态分类

key ed state-这里的键是Gr组/分区组中对应于我们的SQL语句的字段。键值是由groupby/PartitionBy字段组成的行的字节数组。每个密钥都有自己的状态，密钥和密钥之间的状态是不可见的。

OperatorState将用于记录在OperatorState-Flink内部实现源连接器时源数据读取的偏移量。

产能扩张期间的状态再分配

Flink是一个允许大规模有状态流处理的大规模并行分布式系统。为了可伸缩性，Flink作业在逻辑上分解为操作符图，每个操作符的执行在物理上分解为多个并行操作符实例。从概念上来说，Flink中的每个并行操作器实例都是一个独立的任务，可以安排它在自己机器上连接到网络的其他机器上运行。

在Flink的DAG图中，只有边缘连接的节点有网络通信，即整个DAG在垂直方向有网络IO，在水平方向有状态的节点之间没有网络通信，如下图所示。该模型还确保每个操作员实例保持自己的状态，并将其保存在本地磁盘中(远程异步同步)。通过这种设计，任务的所有状态数据都是本地的，并且状态访问不需要任务之间的网络通信。避免这种流量对于像Flink这样的大规模并行分布式系统的可扩展性非常重要。

如上所述，Flink中的State有OperatorState和KeyedState，那么扩展(增加并发)时如何分配状态呢？例如，外部Source有五个分区，从Source的一个并发扩展到Flink上的两个并发，中间的Stateful Operation节点由两个并发和扩展的三个并发组成，如下图所示:

在Flink中，不同类型的状态有不同的扩展方法，接下来我们将分别介绍。

运营商对产能扩张的处理

我们选择了Flink中连接器实现的一个具体例子来介绍。以MetaQ为例。MetaQ按主题订阅数据，每个主题将有N>0个分区。如上图所示，假设我们订阅的MetaQ主题有5个分区，那么当我们的源从一个并发调整到两个并发时，状态会如何恢复呢？

恢复状态的方式与源中操作者状态的存储结构有着必然的关系。让我们首先看看MetaQSource的实现是如何存储状态的。首先，MetaQSource实现ListCheckpointed，其中t是Tuple2。我们看列表检查点接口的内部定义如下:

public interface ListCheckpointed {    List snapshotState(long var1, long var3) throws Exception;    void restoreState(List var1) throws Exception;}

我们发现snapshotState方法的返回值是一个list，t是Tuple2，也就是说snapshotState方法返回一个List。该类型表示状态的存储是包含分区和偏移量信息的列表，InputSplit表示分区，Long表示当前分区读取的偏移量。InputSplit有一种方法如下:

public interface InputSplit extends Serializable {    int getSplitNumber();}

也就是说，InputSplit可以理解为一个分区索引。有了这个数据结构，我们来看看上图所示的案例是如何工作的。当Source的并行度为1时，所有分区数据都在同一个线程中读取，所有分区的状态也保持在同一状态。状态存储信息的格式如下: