Flink系列-第09讲：Flink 状态与容错

这一课时我们主要讲解 Flink 的状态和容错。

在 Flink 的框架中，进行有状态的计算是 Flink 最重要的特性之一。所谓的状态，其实指的是 Flink 程序的中间计算结果。Flink 支持了不同类型的状态，并且针对状态的持久化还提供了专门的机制和状态管理器。

状态

我们在 Flink 的官方博客中找到这样一段话，可以认为这是对状态的定义：

When working with state, it might also be useful to read about Flink’s state backends. Flink provides different state backends that specify how and where state is stored. State can be located on Java’s heap or off-heap. Depending on your state backend, Flink can also manage the state for the application, meaning Flink deals with the memory management (possibly spilling to disk if necessary) to allow applications to hold very large state. State backends can be configured without changing your application logic.

这段话告诉我们，所谓的状态指的是，在流处理过程中那些需要记住的数据，而这些数据既可以包括业务数据，也可以包括元数据。Flink 本身提供了不同的状态管理器来管理状态，并且这个状态可以非常大。

Flink 的状态数据可以存在 JVM 的堆内存或者堆外内存中，当然也可以借助第三方存储，例如 Flink 已经实现的对 RocksDB 支持。Flink 的官网同样给出了适用于状态计算的几种情况：

• When an application searches for certain event patterns, the state will store the sequence of events encountered so far
• When aggregating events per minute/hour/day, the state holds the pending aggregates
• When training a machine learning model over a stream of data points, the state holds the current version of the model parameters
• When historic data needs to be managed, the state allows efficient access to events that occurred in the past

以上四种情况分别是：复杂事件处理获取符合某一特定时间规则的事件、聚合计算、机器学习的模型训练、使用历史的数据进行计算。

Flink 状态分类和使用

我们在之前的课时中提到过 KeyedStream 的概念，并且介绍过 KeyBy 这个算子的使用。在 Flink 中，根据数据集是否按照某一个 Key 进行分区，将状态分为 Keyed State 和 Operator State（Non-Keyed State）两种类型。

如上图所示，Keyed State 是经过分区后的流上状态，每个 Key 都有自己的状态，图中的八边形、圆形和三角形分别管理各自的状态，并且只有指定的 key 才能访问和更新自己对应的状态。

与 Keyed State 不同的是，Operator State 可以用在所有算子上，每个算子子任务或者说每个算子实例共享一个状态，流入这个算子子任务的数据可以访问和更新这个状态。每个算子子任务上的数据共享自己的状态。

但是有一点需要说明的是，无论是 Keyed State 还是 Operator State，Flink 的状态都是基于本地的，即每个算子子任务维护着这个算子子任务对应的状态存储，算子子任务之间的状态不能相互访问。

我们可以看一下 State 的类图，对于 Keyed State，Flink 提供了几种现成的数据结构供我们使用，State 主要有四种实现，分别为 ValueState、MapState、AppendingState 和 ReadOnlyBrodcastState ，其中 AppendingState 又可以细分为 ReducingState、AggregatingState 和 ListState。

那么我们怎么访问这些状态呢？Flink 提供了 StateDesciptor 方法专门用来访问不同的 state，类图如下：

下面演示一下如何使用 StateDesciptor 和 ValueState，代码如下：

public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 5L), Tuple2.of(1L, 2L))
.keyBy(0)
.flatMap(new CountWindowAverage())
.printToErr();
   env.execute(<span class="hljs-string">"submit job"</span>);

}
public static class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
   <span class="hljs-keyword">private</span> <span class="hljs-keyword">transient</span> ValueState&lt;Tuple2&lt;Long, Long&gt;&gt; sum;
   <span class="hljs-function"><span class="hljs-keyword">public</span> <span class="hljs-keyword">void</span> <span class="hljs-title">flatMap</span><span class="hljs-params">(Tuple2&lt;Long, Long&gt; input, Collector&lt;Tuple2&lt;Long, Long&gt;&gt; out)</span> <span class="hljs-keyword">throws</span> Exception </span>{

       Tuple2&lt;Long, Long&gt; currentSum;
       <span class="hljs-comment">// 访问ValueState</span>
       <span class="hljs-keyword">if</span>(sum.value()==<span class="hljs-keyword">null</span>){
           currentSum = Tuple2.of(<span class="hljs-number">0L</span>, <span class="hljs-number">0L</span>);
       }<span class="hljs-keyword">else</span> {
           currentSum = sum.value();
       }

       <span class="hljs-comment">// 更新</span>
       currentSum.f0 += <span class="hljs-number">1</span>;

       <span class="hljs-comment">// 第二个元素加1</span>
       currentSum.f1 += input.f1;

       <span class="hljs-comment">// 更新state</span>
       sum.update(currentSum);

       <span class="hljs-comment">// 如果count的值大于等于2，求知道并清空state</span>
       <span class="hljs-keyword">if</span> (currentSum.f0 &gt;= <span class="hljs-number">2</span>) {
           out.collect(<span class="hljs-keyword">new</span> Tuple2&lt;&gt;(input.f0, currentSum.f1 / currentSum.f0));
           sum.clear();
       }

}
public void open(Configuration config) {
ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
new ValueStateDescriptor<>(
“average”, // state的名字
TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {})
); // 设置默认值
   StateTtlConfig ttlConfig = StateTtlConfig
           .newBuilder(Time.seconds(<span class="hljs-number">10</span>))
           .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
           .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
           .build();

   descriptor.enableTimeToLive(ttlConfig);

   sum = getRuntimeContext().getState(descriptor);
   }

}

在上述例子中，我们通过继承 RichFlatMapFunction 来访问 State，通过 getRuntimeContext().getState(descriptor) 来获取状态的句柄。而真正的访问和更新状态则在 Map 函数中实现。

我们这里的输出条件为，每当第一个元素的和达到二，就把第二个元素的和与第一个元素的和相除，最后输出。我们直接运行，在控制台可以看到结果：

Operator State 的实际应用场景不如 Keyed State 多，一般来说它会被用在 Source 或 Sink 等算子上，用来保存流入数据的偏移量或对输出数据做缓存，以保证 Flink 应用的 Exactly-Once 语义。

同样，我们对于任何状态数据还可以设置它们的过期时间。如果一个状态设置了 TTL，并且已经过期，那么我们之前保存的值就会被清理。

想要使用 TTL，我们需要首先构建一个 StateTtlConfig 配置对象；然后，可以通过传递配置在任何状态描述符中启用 TTL 功能。

StateTtlConfig ttlConfig = StateTtlConfig
        .newBuilder(Time.seconds(10))
        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
        .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
        .build();
descriptor.enableTimeToLive(ttlConfig);

StateTtlConfig 这个类中有一些配置需要我们注意：

UpdateType 表明了过期时间什么时候更新，而对于那些过期的状态，是否还能被访问则取决于 StateVisibility 的配置。

状态后端种类和配置

我们在上面的内容中讲到了 Flink 的状态数据可以存在 JVM 的堆内存或者堆外内存中，当然也可以借助第三方存储。默认情况下，Flink 的状态会保存在 taskmanager 的内存中，Flink 提供了三种可用的状态后端用于在不同情况下进行状态后端的保存。

• MemoryStateBackend
• FsStateBackend
• RocksDBStateBackend

MemoryStateBackend

顾名思义，MemoryStateBackend 将 state 数据存储在内存中，一般用来进行本地调试用，我们在使用 MemoryStateBackend 时需要注意的一些点包括：

每个独立的状态（state）默认限制大小为 5MB，可以通过构造函数增加容量
状态的大小不能超过 akka 的 Framesize 大小
聚合后的状态必须能够放进 JobManager 的内存中

MemoryStateBackend 可以通过在代码中显示指定：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new MemoryStateBackend(DEFAULT_MAX_STATE_SIZE,false));

其中，new MemoryStateBackend(DEFAULT_MAX_STATE_SIZE,false) 中的 false 代表关闭异步快照机制。关于快照，我们在后面的课时中有单独介绍。

很明显 MemoryStateBackend 适用于我们本地调试使用，来记录一些状态很小的 Job 状态信息。

FsStateBackend

FsStateBackend 会把状态数据保存在 TaskManager 的内存中。CheckPoint 时，将状态快照写入到配置的文件系统目录中，少量的元数据信息存储到 JobManager 的内存中。

使用 FsStateBackend 需要我们指定一个文件路径，一般来说是 HDFS 的路径，例如，hdfs://namenode:40010/flink/checkpoints。

我们同样可以在代码中显示指定：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints", false));

FsStateBackend 因为将状态存储在了外部系统如 HDFS 中，所以它适用于大作业、状态较大、全局高可用的那些任务。

RocksDBStateBackend

RocksDBStateBackend 和 FsStateBackend 有一些类似，首先它们都需要一个外部文件存储路径，比如 HDFS 的 hdfs://namenode:40010/flink/checkpoints，此外也适用于大作业、状态较大、全局高可用的那些任务。

但是与 FsStateBackend 不同的是，RocksDBStateBackend 将正在运行中的状态数据保存在 RocksDB 数据库中，RocksDB 数据库默认将数据存储在 TaskManager 运行节点的数据目录下。

这意味着，RocksDBStateBackend 可以存储远超过 FsStateBackend 的状态，可以避免向 FsStateBackend 那样一旦出现状态暴增会导致 OOM，但是因为将状态数据保存在 RocksDB 数据库中，吞吐量会有所下降。

此外，需要注意的是，RocksDBStateBackend 是唯一支持增量快照的状态后端。

总结

我们在这一课时中讲解了 Flink 中的状态分类和使用，并且用实际案例演示了用法；此外介绍了 Flink 状态的保存方式和优缺点。

点击这里下载本课程源码

---

精选评论

*勇：

你好，从kafka消费数据，通过FLINK ETL处理后，保存到HDFS中。现在由于FLINK环境不稳定，经常会重启任务，导致数据丢失。如何处理？

    讲师回复：

    注意哈，Flink是可以保证exactly-once的，你的这种情况，频繁重启，是因为没有进行合理调优。主要从两个方面入手：1.看看写入瓶颈在哪里？2.看看反压情况。要优化你的任务了。

*邦：

想问下,checkpoint恢复是指的是重试次数内的自动故障恢复么。如果这个job因为某种原因完全失败,那再手动提交的时候还能访问之前的checkpoint么。这种情况,是怎么处理。

    讲师回复：

    这里的checkpoint恢复包括任务在运行时遇到的各种问题导致任务失败，会根据配置不断进行重试。如果手动停止，修改代码，重新上线后不会在访问历史checkpoint了，因为这时候你的历史checkpoint的结构和内容全变了。

**昶：

你好,请教一个问题，keyby()后，在RichKeyProcess()中，我用一个布尔型状态记录每个key是否在白名单中，这些状态都会在每天0点清除，如果key的数量很大，那就会产生很多定时器，这些定时器会有什么影响？key的数量达到什么会产生影响，谢谢啦

    讲师回复：

    首先定时器会存储在JVM中，会占用大量内存。如果你的定时器数量过多，当定时器触发时会产生计算风暴，性能低下，需要加大并行度，将定时器分散到不同的task节点上。

**立：

不太明白使用sum.update(currentSum);更新了状态，为何还要调用sum.clear()清除状态

    讲师回复：

    在这里举的例子是大于2后清除状态，实际应用中可以根据业务需要，对状态是否清除自由管理。

*亮：

你好，“每个算子子任务上的数据共享自己的状态。但是有一点需要说明的是，无论是 Keyed State 还是 Operator State，Flink 的状态都是基于本地的，即每个算子子任务维护着这个算子子任务对应的状态存储，算子子任务之间的状态不能相互访问。”这句话意思是状态维护中计算的数据状态是共享的，但是算子本身的运行状态是独立的是吧，这么理解对么？

    讲师回复：

    可以这么理解，每个算子的状态互相独立，不会互相影响。

*熙：

状态，容错，就讲了状态，状态怎么存储，那容错呢？怎么容错的？怎么恢复呢？？？

    讲师回复：

    容错恢复就是从状态、checkpoint进行恢复

**-4年-大数据开发：

rocksdb到底怎么使用，我使用它也会出现oom，他是一个数据库，如果配置的是hdfs，那么hdfs文件就是他的数据库对应的文件是吗？

    讲师回复：

    flink-statebackend-rocksdb_2.11需要依赖这个包。

褚：

实际使用场景是什么呢？

    讲师回复：

    这是Flink底层的实现，在几乎所有实时计算任务下都会使用。