Flink的State扩容机制

何为State

为实现增量计算和容错，Flink提出了State机制，本质上State就是用来存放计算过程中各节点的中间结果或元数据等，并提供Exactly-Once语义。

流计算的大部分场景均是增量计算的，数据逐条被处理，每次当前结果均是基于上一次计算结果之上进行处理的，这势必需要将上一次的计算结果进行存储持久化。

目前Flink有3种State存储实现:

• HeapStateBackend，内存，存放数据量小，用于开发测试，生产环境不建议；
• FsStateBackend，分布式文件系统(HDFS等)，不支持增量，可用于大State，可用于生产环境；
• RockDBStateBackend，RockDB，支持增量，可用于大State，可用于生产环境。

生产环境下的最佳实践为:

2.jpg

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扩容，指的当算子并行度发生改变时，其需要进行相应的组织调整。

如下图所示:

3.jpg

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如下所示:

4.jpg

当Source的并行度修改为2之后，Task与State的对应关系如下:

5.jpg

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个例子来实际说明:

1.jpg

• 当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均不需要重新组织。

核心思想就是: 不直接操作数据，只操作数据的指针。