Flink中的State概念及其扩容算法 state是什么意思 _小知识

在流资源网盘算场景中，数据会源源不断的流入Apache Flink体系，每条数据进入Apache Flink体系都会触发盘算。如果我们想进行一个Count聚合盘算，那么每次触发盘算是将历史上所有流入的数据重新盘算一次，还是每次盘算都是在上一次盘算成果之上进行增量盘算呢?答案是确定的，Apache Flink是基于上一次的盘算成果进行增量盘算的。那么问题来了: "上一次的盘算成果保留在哪里，保留在内存可以吗?"，答案是否认的，如果保留在内存，在由于网络，硬件等原因造成某个盘算节点失败的情形下，上一次盘算成果会丧失，在节点恢复的时候，就须要将历史上所有数据(可能十几天，上百天的数据)重新盘算一次，所认为了避免这种灾害性的问题产生，Apache Flink 会应用State存储盘算成果。本篇将会为大家介绍Apache Flink State的相干内容。

文章插图

什么是State这个问题似乎有些"弱智"？不管问题的答案是否显而易见，但我还是想简略说一下在Flink里面什么是State？State是指流盘算进程中盘算节点的中间盘算成果或元数据属性，比如在aggregation进程中要在state中记载中间聚合成果，比如 Apache Kafka 作为数据源时候，我们也要记载已经读取记载的offset，这些State数据在盘算进程中会进行持久化(插入或更新) 。所以Flink中的State就是与时光相干的，Flink义务的内部数据（盘算数据和元数据属性）的快照。
为什么须要State与批盘算相比，State是流盘算特有的，批盘算没有failover机制，要么胜利，要么重新盘算。流盘算在大多数场景下是增量盘算，数据逐条处置（大多数场景)，每次盘算是在上一次盘算成果之上进行处置的，这样的机制势必要将上一次的盘算成果进行存储（生产模式要持久化），另外由于机器，网络，脏数据等原因导致的程序毛病，在重启job时候须要从胜利的检讨点(checkpoint，后面篇章会专门介绍)进行state的恢复。增量盘算，Failover这些机制都须要state的支持。
State 存储实现Flink内部有三种state的存储实现，具体如下：

基于内存的HeapStateBackend - 在debug模式应用，不建议在生产模式下运用；
基于HDFS的FsStateBackend - 散布式文件持久化，每次读写都操作内存，同需斟酌OOM问题；
基于RocksDB的RocksDBStateBackend - 本地文件+异步HDFS持久化;

State存储的架构Apache Flink 默认是RocksDB+HDFS的方法进行State的存储，State存储分两个阶段，首先本地存储到RocksDB，然后异步的同步到远程的HDFS 。这样的而设计既清除了HeapStateBackend的局限（内存大小，机器坏掉丧失等），也减少了纯散布式存储的网络IO开销。

文章插图

State 分类

KeyedState - 这里面的key是我们在SQL语句中对应的GroupBy/PartitioneBy里面的字段，key的值就是groupby/PartitionBy字段组成的Row的字节数组，每一个key都有一个属于自己的State，key与key之间的State是不可见的；
OperatorState - Flink内部的Source Connector的实现中就会用OperatorState来记载source数据读取的offset 。

State在扩容时候的重新分配Flink是一个大范围并行散布式体系，许可大范围的有状况流处置。资源网为了可伸缩性，Flink作业在逻辑上被分解成operator graph，并且每个operator的履行被物理地分解成多个并行运算符实例。从概念上讲，Flink中的每个并行运算符实例都是一个独立的义务，可以在自己的机器上调度到网络衔接的其他机器运行。
Flink的DAG图中只有边相连的节点有网络通讯，也就全部DAG在垂直方向有网络IO，在程度方向如下图的stateful节点之间没有网络通讯，这种模型也保证了每个operator实例保护一份自己的state，并且保留在本地磁盘（远程异步同步）。通过这种设计，义务的所有状况数据都是本地的，并且状况拜访不须要义务之间的网络通讯。避免这种流量对于像Flink这样的大范围并行散布式体系的可扩大性至关主要。
如上我们知道Flink中State有OperatorState和KeyedState，那么在进行扩容时候（增长并发）State如何分配呢？比如：外部Source有5个partition，在Flink上面由Source的1个并发扩容到2个并发，中间Stateful Operation 节点由2个并发并扩容的3个并发，如下图所示: