何为State
为实现增量计算和容错，Flink提出了State机制，本质上State就是用来存放计算过程中各节点的中间结果或元数据等，并提供Exactly-Once语义。
流计算的大部分场景均是增量计算的，数据逐条被处理，每次当前结果均是基于上一次计算结果之上进行处理的，这势必需要将上一次的计算结果进行存储持久化。
目前Flink有3种State存储实现:
- HeapStateBackend，内存，存放数据量小，用于开发测试，生产环境不建议；
- FsStateBackend，分布式文件系统(HDFS等)，不支持增量，可用于大State，可用于生产环境；
- RockDBStateBackend，RockDB，支持增量，可用于大State，可用于生产环境。
生产环境下的最佳实践为:
State先在本地存储到RockDB，然后异步写入到HDFS，即避免了HeapStateBackend的单节点资源限制(物理内存、机器宕机丢失数据等)，也减少了分布式写入带来的网络IO开销。
State分类
从Operator和Data角度可将State分为2类:
- OperatorState：Source Connector的实现中就会用OperatorState来记录source数据读取的offset。
- KeyedState：groupby或partitionBy组成的内容，即为key，每个key均有自己的state，且key与key之间的state不可见。
State扩容
所谓State扩容，指的当算子并行度发生改变时，其需要进行相应的组织调整。
如下图所示:
OperatorState扩容
OperatorState往往以ListState<T>的形式存在，如FlinkKafkaConsumerBase:
@Internal
public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFunction<T> implements
CheckpointListener,
ResultTypeQueryable<T>,
CheckpointedFunction {
...
/** Accessor for state in the operator state backend. */
private transient ListState<Tuple2<KafkaTopicPartition, Long>> unionOffsetStates;
...
}
此时，T对应Tuple2<KafkaTopicPartition, Long>，KafkaTopicPartition代表Kafka Topic的一个Partition，Long代表当前Partition的offset。
假设某Topic的partition数目为5，且Source的并行度=1，则对应的State如下所示:
当Source的并行度修改为2之后，Task与State的对应关系如下:
KeyedState扩容
Flink Source执行keyBy之后，各个元素会基于key链接到下游不同的并行Operator上，流计算中同时会涉及到KeyedState的组织。
key的数目一般大于Operator的并行度parallelism，最直观的做法是将key的hash值与并行度parallelism取余。
假设上游10个元素，其keyHash分别为{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}，下游Operator的并行度parallelism为2。
- operatorIndex=0，分配到的元素有0, 2, 4, 6, 8，维护的State-0值为0, 2, 4, 6, 8
- operatorIndex=1，分配到的元素有1, 3, 5, 7, 9，维护的State-1值为1, 3, 5, 7, 9
假设下游Operator的并行度parallelism为修改为3，此时:
- operatorIndex=0，分配到的元素有0, 3, 6, 9，维护的State-0值为0, 3, 6, 9
- operatorIndex=1，分配到的元素有1, 4, 7，维护的State-1值为1, 4, 7
- operatorIndex=2，分配到的元素有2, 5, 8，维护的State-2值为2, 5, 8
假如并行度parallelism发生改变的话，则前面维护好的State也需要重新组织一遍。KeyedState数据较大时，数据重新组织的代价较高。
为了解决上述问题，Flink采用了一种KeyGroupRange的机制，基本思想是将各元素先分配到最细粒度的组中，Flink将其称为KeyGroup，KeyGroup也是KeyedState的最小组织单位。然后并行Operator持有各自的KeyGroup集合即可，该集合即所谓的KeyGroupRange。
public class KeyGroupRange implements KeyGroupsList, Serializable {
...
private final int startKeyGroup;
private final int endKeyGroup;
...
}
很明显，其通过一个范围来定义集合，范围起点为startKeyGroup，终点为endKeyGroup，左闭右闭。
我们知道，各个算子均有最大并行度maxParallelism，所以可以利用key的hash值与maxParallelism进行取模来完成KeyGroup的构建。
public static int assignToKeyGroup(Object key, int maxParallelism) {
return computeKeyGroupForKeyHash(key.hashCode(), maxParallelism);
}
public static int computeKeyGroupForKeyHash(int keyHash, int maxParallelism) {
return MathUtils.murmurHash(keyHash) % maxParallelism;
}
Flink没有直接使用hashcode，而是在hashcode的基础上又调用了murmurHash方法，以保证尽量的散列。
现在有maxParallelism个KeyGroup，需要将其分配到parallelism个并行算子中，每个并行算子持有1个KeyGroupRange，其起终点的计算方式如下:
public static KeyGroupRange computeKeyGroupRangeForOperatorIndex(
int maxParallelism,
int parallelism,
int operatorIndex) {
checkParallelismPreconditions(parallelism);
checkParallelismPreconditions(maxParallelism);
Preconditions.checkArgument(maxParallelism >= parallelism,
"Maximum parallelism must not be smaller than parallelism.");
int start = ((operatorIndex * maxParallelism + parallelism - 1) / parallelism);
int end = ((operatorIndex + 1) * maxParallelism - 1) / parallelism;
return new KeyGroupRange(start, end);
}
最后，每个元素对应的算子计算方式如下:
public static int assignKeyToParallelOperator(Object key, int maxParallelism, int parallelism) {
return computeOperatorIndexForKeyGroup(maxParallelism, parallelism, assignToKeyGroup(key, maxParallelism));
}
public static int computeOperatorIndexForKeyGroup(int maxParallelism, int parallelism, int keyGroupId) {
return keyGroupId * parallelism / maxParallelism;
}
上面说的可能还是有点抽象，下面用1个例子来实际说明:
- 当parallelism=2时可得到KeyGroupRange：
operatorIndex=0，则得到start=0，end=4:如图kg-keys：0，1，2，3，4
operatorIndex=1，则得到start=5，end=9:如图kg-keys：5，6，7，8，9
- 当parallelism=3时可得到KeyGroupRange：
operatorIndex=0，则得到start=0，end=3:如图kg-keys：0，1，2，3
operatorIndex=1，则得到start=4，end=6:如图kg-keys：4，5，6
operatorIndex=2，则得到start=7，end=9:如图kg-keys：7，8，9
采用KeyGroupRange机制，只要Flink任务的maxParallelism配置不变，无论算子的parallelism如何变化，底层的KeyedSate均不需要重新组织。
核心思想就是: 不直接操作数据，只操作数据的指针。
