流式计算系统系列：总纲​zhuanlan.zhihu.com

状态需求的来源

现实中的计算可以根据对状态的需求分为无状态的计算和有状态的计算两种。无状态的计算意味着单条输入包含计算出输出的所有信息，不需要其他信息即可得到输出；而有状态的计算意味着单条输入仅包含计算出输出的部分信息，其他信息依赖于之前的输入积累的状态。无状态的计算典型的例子比如从 socket 源将数据导入 HDFS，有状态的计算典型的例子比如经典的 WordCount 就需要累积之前的单词计数。

显而易见，大多数有业务价值的计算场景都是有状态的计算。状态的需求包括状态数据的存储和加载，快照和恢复，以及分布式场景下的划分和伸缩三个方面。批处理和流式计算都有对状态的需求，但是其重要性和实际处理方式并不相同。

对于状态数据的存储和加载，在批处理的场景下，单条输入实际上意味着单批数据。数据被划分为不同的分片，通常，根据具体的操作（如 KeyBy）被划分为彼此相关的若干块。在这种划分方式下，每个任务处理其中一个分片，最终聚合输出得到结果。在这个过程中，相互依赖的数据被分在一起，并且每个批次都只有有限的数据，内存状态即可解决有状态计算中数据相互依赖的问题，对于状态的需求并不明显。

但在流式计算中，单条输入就意味着单条数据。由于流式计算的输入理论上是个无限流，我们必须有某种方式来管理先前输入积累的状态。注意到区别于批处理分钟、小时级的处理时间，流式计算的运行时间以天、月甚至年为单位。在这种情况下，我们不能总是依赖内存来管理状态。内存状态的管理方式并不能很好的应对大数据量的状态的存储问题，容错场景下的一致性问题和横向扩展时的伸缩问题。

存储的问题是明显的，批处理每批只处理有限的数据，而流式计算需要应对无限的输入。容错场景下的一致性问题在本系列第一篇文章中也有提到，在批处理的场景下，单批失败只需要重新计算，而流式计算的场景里上游无法无限缓存中间结果，必须依赖状态的快照和恢复来容错。如果说在批处理的场景下，快照是一个快速恢复的优化，那么在流式计算的场景里 ，这就是一个必须实现的功能了。

另外，状态的划分和应对规模伸缩在流式计算中也需要被特别对待。这是因为在批处理中，状态是和一批有限的数据绑定的，在规模伸缩时，只要把数据按照不同的方式重新划分好，状态自然就跟着数据被划分好了，也就是说，状态是跟着有限的数据走的。对于已完成的计算如果牵扯进状态划分，则可以重算一遍有限数据的批次。但是在流式计算中，我们不可能实际的完整划分无限的数据，也不可能回溯过往的所有数据，规模伸缩实际改变的是网络拓扑，也就是所谓的重分区（repartition），见下图。在这种情况下，我们必须做好状态数据的格式定义，以在重分区的场景下正确地将状态重新划分。当然，批处理也可以做这样的优化避免已计算完的状态重新计算，不过只是一个优化，而且对于正在执行的作业，这样的优化有些没必要。

传统的流式计算对状态管理仅提供有限的支持（Spark Streaming）或者根本不提供支持（Storm），但是有状态的计算是确实存在的业务需求。在这种情况下采取的方法往往是引入另一个系统来辅助实现状态管理。例如经典的 Storm + HBase 架构，把状态数据存放在 HBase 中，计算的时候再从 HBase 中读取状态数据。但是这样的方案显然有不可忽略的性能开销，而且仅仅是解决了流式计算可能产生超大数据量的问题，以及持久化容错但不保证强一致性。同时，这样的方案在伸缩的时候需要用户手动编写代码去重新分配状态，更广泛的说，每一个业务自己都要重新实现一整套状态管理的逻辑。这种重复实现不仅是没必要的，而且显而易见的会在每次实现的时候碰到相同不相同的 BUG。

我们不会要求业务从操作系统开始自己写起来搭建应用，作为流式计算的操作系统，其本身也应该支持这种常见而且有规律的需求，即提供开箱即用的本地状态管理。这里的本地是相对于 Storm + HBase 方案中需要对每个操作都走网络而言的，强调的是状态和计算在同一台机器上，包括但不限于内存状态。

状态种类的划分

新时代的流式计算系统基本都将状态管理考虑在需要支持的核心特性当中，其中 Flink 由于长时间的考验和眼睛，支持的状态种类和兼顾到的应用场景是最多的。我们来看到 Flink 中的状态种类划分，其他的系统基本大同小异。

总的来说，Flink 的状态区分为 KeyedState 和 OperatorState 两种。顾名思义，KeyedState 跟每一个 key 相关联，也就只能使用在 KeyedStream 当中；而 OperatorState 与每个算子相关，可以使用在任意算子里。

首先说 KeyedState，我们先看一段 Flink 代码，感受一下实际使用的情况。

// KeyedState
public class WordCount extends RichFlatMapFunction<String, Tuple2<String, Integer>> {
  private transient ValueState<Integer> state;

  @Override
  public void open(Configuration conf) {
    final ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>("C", Integer.class, );
    state = getRuntimeContext().getState(descriptor);
  }

  @Override
  public void flatMap(String word, Collector<Tuple2<String, Integer>> collector) {
    int c = state.value();
    c += 1;
    state.update(c);
    collector.collect(new Tuple2<>(word, count));
  }
}

状态管理的架构和控制流会在下一节展开，这里我们关注到状态的具体使用。我们在算子的一个并发启动的时候初始化状态的句柄，随后在实际计算的过程中通过句柄来获取和更新算子。在背后，Flink 通过后文提到的 StateBackend 等抽象完成了实际读写内存或者加载存储持久化后端以及可能的序列化操作。对于用户来说，这些都是不可知的，开箱即用的。

这里我们看到使用了名为 ValueState 的状态，顾名思义，状态就像一个值一样可以被读取和更新。Flink 的 KeyedState 还支持 MapState/ListState/ReducingState/AggregatingState，其中 Map 和 List 好理解，和 Value 可以某种程度上组成 Perl 三巨头类型。后两个 State 其实也是某种意义上的 ListState，只不过 Flink 为未来也要进行状态合并的状态提供了提前压缩状态的优化，以减少状态的大小并分摊计算压力。

然后我们看到 OperatorState，同样先看一段使用代码。

// OperatorState
public class MapF<T> extends RichMapFunction<T, T> implements CheckpointedFunction {
  private transient ListState<Long> countState;
  private transient List<Long> count;

  @Override
  public void initializeState(FunctionInitializationContext context) throws Exception {
    ListStateDescriptor<List> descriptor = new ListStateDescriptor<>("es", Long.class);
    countState = context.getOperatorStateStore().getListState(descriptor);
  }

  @Override
  public void snapshotState(FunctionSnapshotContext context) throws Exception {
    countState.clear();
    countState.addAll(count);
  }
}

可以看到，对于 OperatorState，用户需要手动管理 restore 和 snapshot 的逻辑。并且我们这里使用了 ListState 来实现类似于 ValueState 的功能，这是因为 OperatorState 仅支持少数几种类型的状态的原因。

那么，KeyedState 和 OperatorState 为什么有这么多不同呢？对于这个问题，我们就要从 KeyedState 和 OperatorState 最初引入所要解决的问题来说明了。

虽然首先引入的是 KeyedState，但是更广泛和通用的概念却是 OperatorState。按照我们一般的想法，状态类似于一个持久化可容错的本地存储，看起来应该跟一个内存变量没有太多的区别才对，按照 Operator 划分，在算子的每一份并发里交流是很正常的事情。实际情况确实大抵也是如此，如果你没有重伸缩（rescale）的需求，把 ListState 在其他场景下当做单元素的列表用，基本也能搞定大多数的需求，对于原先 KeyedState 的场景，只要手动加一个 key 到 value 的映射就可以了。

那么，问题就出在重伸缩上面。正如我们在状态需求的来源一节中提到的，好的状态设计要能够应对规模伸缩，如前图所示，当某个算子的并发度从 3 改成 2 时，算子的状态并不好简单的划分。实际上，前文提到的 OperatorState 仅支持少数几种类型的状态，这些状态不同之处主要就在于重新划分状态时的策略。我们先看看都是什么策略，然后再引入 KeyedState 处理这个问题的方法。

OperatorState 支持 SPLIT_DISTRIBUTE(ListState)/UNION(UnionListState)/BROADCAST(BroadcastState) 三种类型，它们在改变并发的时候重新划分的逻辑如下图所示。

• ListState: 把所有状态先后串接在一起，然后均匀分配到新的并发上。
• UnionListState: 把所有状态先后串接在一起，然后将串接后的全量分配到新的并发上。
• BroadcastState: 此状态本来每份并发就都会拥有同样的全量数据，直接拷贝到新的并发上即可。

由于 OperatorState 并不知道用户划分状态的依据，因此只能提供这样有限的接口，对于上面使用的 count 例子，由于有交换律和结合律，还可直接使用 ListState，而非这种情况也非 broadcase 情况的，则算子不得不全量获取状态并由用户在 restore 里重新实现具体使用哪部分状态的逻辑。这个重新划分的逻辑并不好做，因为状态作为过往输入综合影响的结果，在重伸缩的情景下，上游算子的输出会发到下游算子的哪个并发上实际是由分区逻辑决定的，在用户层面，很难直接取得网络拓扑的信息，用户也不应该直接分析网络拓扑这样底层的逻辑。无论是由划分决定网络，还是本质上的网络决定划分，OperatorState 在更一般的场景下都要求更底层的操作状态，而这正是我们提供开箱即用的状态管理想要避免的问题。

KeyedState 就很好的解决了这个问题。在许多实际的业务场景里，数据并不是杂乱无章的被处理，而是有各种各样的标签信息。例如单词计数中的词汇和当前计数，以及用户点击中的用户名和点击行为。在这些场景中，我们会希望将数据按照一定的信息划分归类，在这里我们就引入了用户划分的逻辑。既然知道了用户划分的逻辑，重伸缩的时候划分状态也就好办了，只要保证原来在同一个 key 下的状态依然被划分到同一个并发上就可以了。如下图所示，其中不同颜色代表不同 key 的状态，状态有大有小。

从概念上说，你可以给数据都加上一个唯一的 id 作为 key，从而利用 KeyedState 进行状态存储，这比起 OperatorState 来也不会缺少任何功能，同时还能很好的支持重伸缩下的重分区操作。那为什么我们还需要 OperatorState 呢？在实际应用场景中，OperatorState 主要用于 Source 的状态管理，例如 KafkaConsumer，此时数据还未流经流式计算引擎，自然也不可能有什么流式计算引擎认识的 key，同时多数的 Source 支持按 offset 回拨，而 offset 本身在 Source 的哪个并发并不重要，可以适应 OperatorState 中 ListState 的场景。另外，在 BroadcastState 的场景里，我们需要把相同的数据发送到每一个下游算子上，这个 KeyedState 反而在重分区的时候不好操作。

状态管理的架构

状态管理的架构根据不同系统的具体实现会有很不一样的表现。例如 Flink 只是一个计算引擎，它会引入 StateBackend 的概念来支持不同的状态持久化存储后端，而 Hazelcast Jet 则由于构建在分布式网格存储 Hazelcast 之上，因此所有的状态都是存储在 Hazelcast 上的。本节的内容会偏向于介绍 Flink 选择的状态管理架构，不同于用户界面的大同小异，封装在后端的状态管理架构跟具体的项目实现相关。

我们不会去讲太多实际代码的细节，因为 Flink 在状态管理这边的实现命名比较混乱，设计比较随意，各种莫名的中间层也很多。我们抓住链路中关键的几个节点，根据 Flink 中状态最主要的分类 KeyedState 和 OperatorState 简要地展开介绍。注意这里讲的是状态管理的架构，不会过多涉及到 Checkpoint 的机制，这两者是相关联但是可以分开看到的功能组件。

Flink 状态管理架构的枢纽是 StateBackend，准确来说，是实际实现的 KeyedStateBackend 和 OperatorStateBackend，它们向上承接用户界面的操作，向下处理具体状态后端的输出工作。

其中，KeyedStateBackend 支持将状态存储到内存中，或者文件系统上，或者 RocksDB 里；而 OperatorStateBackend 则只支持内存和文件系统的状态后端。这是官方的和常用的说法，但其实这里统称为状态后端的概念有两个不同维度的区分。

第一个是 Working State 的位置，也就是用户函数里实际操作的 State 保存的位置，这个要么是在内存里，要么是在 RocksDB 里，其中 RocksDB 作为更大但是稍慢的内存来使用，同时利用 RocksDB 的存储格式，可以适当地压缩状态数据并支持增量的 Checkpoint。这里提到了 Checkpoint，第二个维度就是 Checkpoint 的位置，也就是状态快照保存的位置，这个要么随着 RPC 信息发送到 JobManager 端保存在内存中，要么就持久化到文件系统里。通常所说的内存状态后端和文件系统状态后端的主要区别即在于此，而 RocksDB 状态后端则可以灵活配置 Checkpoint 持久化的位置，不过基于其一般用于保存大的状态，通常是搭配文件系统状态后端来使用。

KeyedStateBackend 和 OperatorStateBackend 实际控制了状态读写的逻辑。对于 KeyedStateBackend 来说，其中基于 Heap 的实现会在内存中创建 StateTable，而 KeyedStateBackend 基于 RocksDB 的实现则直接使用 RocksDB 的数据库接口来查询和更新。对于 OperatorStateBackend 的唯一实现，则是相应创建 List 的封装或者 Map 封装的 BroadcastState。值得一提的是，KeyedState 包括一个 Namespace 属性，这个玩意主要是给 Window 算子用的，也就是在 Keyed 的第一层划分之后，新加了第二层划分的逻辑。通过将窗口对象设置为 Namespace，就可以将状态先按照 Keyed，再按照 Window 进行两阶段划分。同时在存取的时候将两个划分标准同时作为读取的依据，唯一锁定一份状态。