点一下关注吧！！！非常感谢！！持续更新！！！
目前已经更新到了：
Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（已更完）
Flink（正在更新！）
章节内容
上节我们完成了如下的内容：
Flink 广播状态
基本概念、代码案例、测试结果
状态存储
Flink的一个重要特性就是有状态计算（stateful processing），Flink提供了简单易用的API来存储和获取状态，但是我们还是要理解API背后的原理，才能更好的使用。
State存储方式
Flink为State提供了三种开箱即用的后端存储方式（state backend）：
Memory State Backend
File System （FS）State Backend
RocksDB State Backend
MemoryStateBackend
MemoryStateBackend将工作状态数据保存在TaskManager的Java内存中。Key/Value状态和Window算子使用哈希表存储数值和触发器。进行快照时（CheckPointing），生成的快照数据将和 CheckPoint ACK消息一起发送给 JobManager，JobManager将收到的所有快照数据保存在Java内存中。
MemoryStateBackend现在被默认配置成异步的，这样避免阻塞主线程的pipline处理，MemoryStateBackend的状态存取的数据都非常快，但是不适合生产环境中使用。这是以为它有以下限制：
每个state的默认大小被限制为5MB（这个值可以通过MemoryStateBackend构造方法设置）
每个Task的所有State数据（一个Task可能包含一个Pipline中的多个的Operator）大小不能超过RPC系统的帧大小（akk.framesize 默认10MB）
JobManager收到的State数据总和不能超过JobManager内存
MemoryStateBackend适合的场景：
本地开发和调试
状态很小的作业
FsStateBackend
FsStateBackend 需要配置一个CheckPoint路径，例如：hdfs://xxxxxxx 或者 file:///xxxxx，我们一般都会配置HDFS的目录。
FsStateBackend将工作状态数据保存在TaskManager的Java内存中，进行快照时，再将快照数据写入上面的配置的路径，然后将写入的文件路径告知JobManager。JobManager中保存所有状态的元数据信息（在HA的模式下，元数据会写入CheckPoint目录）。
FsStateBackend 默认使用异步方式进行快照，防止阻塞主线程的Pipline处理，可以通过
FsStateBackend构造函数取消该模式：
new FsStateBackend(path, false)
FsStateBackend 适合的场景：
- 大状态、长窗口、大键值（键或者值很大）状态的作业
- 适合高可用方案
RocksDBStateBackend
RocksDBStateBackend 也需要配置一个CheckPoint路径，例如：hdfs://xxx 或者 file:///xxx，一般是 HDFS 路径。
RocksDB是一种可嵌入的可持久型的 key-value 存储引擎，提供 ACID 支持。由Facebook基于LevelDB开发，使用LSM存储引擎，是内存和磁盘的混合存储。
RocksDBStateBackend将工作状态保存在TaskManager的RocksDB数据库中，CheckPoint时，RocksDB中的所有数据会被传输到配置的文件目录，少量元数据信息保存在JobManager内存中（HA模式下，会保存在CheckPoint目录）。
RocksDBStateBackend使用异步方式进行快照，RocksDBStateBackend的限制：
由于RocksDB的JNI Bridge API是基于 byte[] 的，RocksDBStateBackend支持的每个Key或者每个Value的最大值不超过 2的31次方（2GB）
要注意的是，有merge操作的状态（例如：ListState），可能会在运行过程中超过2的31次时，导致程序失败。
RocksDBStateBackend适用于以下的场景：
超大状态、超长窗口（天）、大键值状态的作业
适合高可用模式
使用RocksDBStateBackend时，能够限制状态大小是TaskManager磁盘空间（相对于FsStateBackend状态大小限制与TaskManager内存）。这也导致RocksDBStateBackend的吞吐比其他两个要低一些，因为RocksDB的状态数据的读写都要经过反序列化/序列化。
RocksDBStateBackend时目前三者中唯一支持增量CheckPoint的。
三者吞吐量对比
KeyedState 和 Operator State
State分类
Operator State
（或 non-keyed state）：
每个Operator State绑定一个并行的Operator实例，KafkaConnector是使用OperatorState的典型示例：每个并行的Kafka Consumer实例维护了每个Kafka Topic分区和该分区Offset的映射关系，并将这个映射关系保存为OperatorState。
在算子并行度改变时，OperatorState也会重新分配。
Keyed State
这种State只存在于KeyedStream上的函数和操作中，比如Keyed UDF（KeyedProcessFunction）Window State。可以把Keyed State想象成被分区的OperatorState。每个KeyedState在逻辑上可以看成与一个 <parallel-operator-instance, key> 绑定，由于一个key肯定只存在于一个Operator实例，所以我们可以简单的的认为一个 <operator, key>对应一个 KeyedState。
每个KeyedState在逻辑上还会被分配一个KeyGroup，分配方法如下：
MathUtils.murmurHash(key.hashCode()) % maxParallelism;
其中maxParallelism是Flink程序的最大并行度，这个值一般我们不会去手动设置，使用默认的值（128）就好，这里注意下，maxParallelism和我们运行程序时指定的算子并行度（parallelism）不同，parallelism不能大于maxParallelism，最多两者相等。
为什么会有 Key Group这个概念呢？
我们通常写程序，会给算子指定一个并行度，运行一段时间后，积累了一些State，这时候数据量大了，需要增加并行度。我们修改并行度后重新提交，那这些已经存在的State该如何分配到各个Operator呢？这就有了最大并行度和KeyGroup的概念。
上面的计算公式也说明了KeyGroup的个数最多是maxParallelism个。当并行度改变之后，我们在计算
这个Key被分配到的Operator：
keyGroupId * paralleism / maxParallelism;
可以看到，一个KeyGroupId会对应一个Operator，当并行度更改时，新的Operator会去拉取对应的KeyGroup的KeyedState，这样就把KeyedState尽量均匀的分配给所有的Operator了。
根据State数据是否被Flink托管，Flink又将State分类为：
Managed State：被Flink托管，保存为内部的哈希表或者RocksDB，CheckPoint时，Flink将State进行序列化编码。例如：ValueState ListState
Row State：Operator 自行管理的数据结构， Checkpoint时，它们只能以byte数据写入CheckPoint。
建议使用 Managed State，当使用 Managed State时，Flink会帮助我们更改并行度时重新分配State，优化内存。
使用ManageKeyedState
如何创建？
上面提到，KeyedState只能在KeyedStream上使用，可以通过Stream.keyBy创建KeyedStream，我们可以创建以下几种：
ValueState
ListState
ReducingState
AggregatingState<IN,OUT>
MapState<UK,UV>
FoldingState<T,ACC>
每种State都对应各种的描述符，通过描述符RuntimeContext中获取对应的State，而RuntimeContext只有RichFunction才能获取，所以想要使用KeyedState，用户编写的类必须继承RichFunction或者其他子类。
ValueState getState(ValueStateDescriptor)
ReducingState getReducingState(ReducingStateDescriptor)
ListState getListState(ListStateDescriptor)
AggregationState<IN,OUT> getAggregatingState(AggregatingStateDescriptor<IN,ACC,OUT>)
FoldingState<T,ACC> getFoldingState(FoldingStateDescriptor<T,ACC>)
MapState<UK,UV> getMapState(MapStateDescriptor<UK,UV>)
给KeyedState设置过期时间
在Flink 1.6.0 以后，还可以给KeyedState设置 TTL（Time-To-Live），当某一个Key的State数据过期时，会被StateBackend尽力删除。
官方给出了示例：
import org.apache.flink.api.common.state.StateTtlConfig; import org.apache.flink.api.common.state.ValueStateDescriptor; import org.apache.flink.api.common.time.Time; StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Time.seconds(1)) // 状态存活时间 .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) // TTL 何时被更新，这里配置的 state 创建和写入时 .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build();// 设置过期的 state 不被读取 ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("textstate", String.class); stateDescriptor.enableTimeToLive(ttlConfig);
State的TTL何时被更新？
可以进行以下配置，默认只是key的state被modify（创建或者更新）的时候才更新TTL：
StateTtlConfig.UpdateType.OnCreateAndWrite：只在一个key的state创建和写入时更新TTL（默认）
StateTtlConfig.UpdateType.onReadAndWrite：读取state时仍然更新TTL
当State过期但是还未删除时，这个状态是否还可见？
可以进行以下配置，默认是不可见的：
StateTtlConfig.StateVisibility.NerverReturnExpired：不可见（默认）
StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp：可见
注意：
状态的最新访问时会和状态数据保存在一起，所以开启TTL特性增大State的大小，Heap State Backend会额外存储一个包括用户状态以及时间戳的Java8对象，RocksDB StateBackend会在每个状态值（list或者map的每个元素）序列化后增加8个字节。
暂时只支持基于 Processing Time的TTL。
尝试从CheckPoint/SavePoint进行恢复时，TTL的状态（是否开启）必须和之前保存一致，否则会遇到：StateMigrationException。
TTL的配置并不会保存在CheckPoint/SavePoint中，仅对当前的Job有效。
当前开启TTL的MapState仅在用户序列化支持NULL的情况下，才支持用户值为NULL，如果用户值序列化器不支持NULL，可以用NullableSerializer包装一层。
使用ManageOperatorState
（这里以及后续放到下一篇：大数据-127 Flink）
