在阅读本文之前,你应该阅读过的系列:
在 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 的官网同样给出了适用于状态计算的几种情况:
以上四种情况分别是:复杂事件处理获取符合某一特定时间规则的事件、聚合计算、机器学习的模型训练、使用历史的数据进行计算。
我们在之前的课时中提到过 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("submit job");
}
public static class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {
private transient ValueState<Tuple2<Long, Long>> sum;
public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {
Tuple2<Long, Long> currentSum;
// 访问ValueState
if(sum.value()==null){
currentSum = Tuple2.of(0L, 0L);
}else {
currentSum = sum.value();
}
// 更新
currentSum.f0 += 1;
// 第二个元素加1
currentSum.f1 += input.f1;
// 更新state
sum.update(currentSum);
// 如果count的值大于等于2,求知道并清空state
if (currentSum.f0 >= 2) {
out.collect(new Tuple2<>(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(10))
.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 将 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 会把状态数据保存在 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 和 FsStateBackend 有一些类似,首先它们都需要一个外部文件存储路径,比如 HDFS 的 hdfs://namenode:40010/flink/checkpoints,此外也适用于大作业、状态较大、全局高可用的那些任务。
但是与 FsStateBackend 不同的是,RocksDBStateBackend 将正在运行中的状态数据保存在 RocksDB 数据库中,RocksDB 数据库默认将数据存储在 TaskManager 运行节点的数据目录下。
这意味着,RocksDBStateBackend 可以存储远超过 FsStateBackend 的状态,可以避免向 FsStateBackend 那样一旦出现状态暴增会导致 OOM,但是因为将状态数据保存在 RocksDB 数据库中,吞吐量会有所下降。
此外,需要注意的是,RocksDBStateBackend 是唯一支持增量快照的状态后端。
Flink中基于异步轻量级的分布式快照技术提供了Checkpoints容错机制,Checkpoints可以将同一时间点作业/算子的状态数据全局统一快照处理,包括前面提到的算子状态和键值分区状态。当发生了故障后,Flink会将所有任务的状态恢复至最后一次Checkpoint中的状态,并从那里重新开始执行。
那么Checkpoints的生成策略是什么样的呢?它会在什么时候进行快照的生成呢?
其实就是在所有任务都处理完同一个输入数据流的时候,这时就会对当前全部任务的状态进行一个拷贝,生成Checkpoints。
为了方便理解,这里先简单的用一个朴素算法来解释这一生成过程(Flink的Checkpoints算法实际要更加复杂,在下面会详细讲解)
为了方便进行实例的讲解,假设当前有一个Source任务,负责从一个递增的数字流(1、2、3、4……)中读取数据,读取到的数据会分为奇数流和偶数流,求和算子的两个任务会分别对它们进行求和。在当前任务中,数据源算子的任务会将输入流的当前偏移量存为状态,求和算子的任务会将当前和存为状态。
某流式应用的一致性检查点
如上图,在当前生成的Checkpoints中保存的输入偏移为5,偶数求和为6,奇数求和为9。
故障:任务sum_odd失败
假设在下一轮计算中,任务sum_odd计算出现了问题,任务sum_odd的时候产生了问题,导致结果出现错误。由于出现问题,为了防止从头开始重复计算,此时会通过Checkpoints来进行快照的恢复。
Checkpoints恢复应用需要以下三个步骤:
恢复步骤1:重启应用
恢复步骤2:从检查点重置任务状态
恢复步骤3:继续处理
从检查点恢复后,它的内部状态会和生成检查点的时候完全一致,并且会紧接着重新处理那些从之前检查点完成开始,到发生系统故障之间已经处理过的数据。虽然这意味着Flink会重复处理部分消息,但上述机制仍然可以实现精确一次的状态一致性,因为所有的算子都会恢复到那些数据处理之前的时间点。
但这个机制仍然面临一些问题,因为Checkpoints和恢复机制仅能重置应用内部的状态,而应用所使用的Sink可能在恢复期间将结果向下游系统(如事件日志系统、文件系统或数据库)重复发送多次。为了解决这个问题,对于某些存储系统,Flink提供的Sink函数支持精确一次输出(在检查点完成后才会把写出的记录正式提交)。另一种方法则是适用于大多数存储系统的幂等更新。
Flink中的Checkpoints是基于Chandy-Lamport分布式快照算法实现的,该算法不会暂停整个应用,而是会将生成Checkpoints的过程和处理过程分离,这样在部分任务持久化状态的过程中,其他任务还可以继续执行。
在介绍生成策略之前,首先需要介绍一下Checkpoints barrier(屏障)
这一种特殊记录。
barrir划分Checkpoints
如上图,与水位线相同,Flink会在Source中间隔性地生成barrier,通过barrier把一条流上的数据划分到不同的Checkpoints中,在barrier之前到来的数据导致的状态更改,都会被包含在当前所属的Checkpoints中;而基于barrier之后的数据导致的所有更改,就会被包含在之后的Checkpoints中。
拥有两个有状态的Source,两个有状态的任务,以及两个无状态Sink的流式应用
JobManager通过向所有Source发送消息来启动Checkpoints生成流
Source为状态生成Checkpoints并发出Checkpoints barrier
任务等待接受所有输入分区的barrier,来自己接受barrier输入分区的记录会被缓存,其他记录按照常规处理
任务在收到全部barrier后将状态存入Checkpoints,然后向下游转发Checkpoints barrier
任务中收齐全部输入分区发送的barrier后,就会通知状态后端开始生成Checkpoints,同时继续把Checkpoints barrier广播转发到下游相连的任务。
任务在转发Checkpoints barrier后继续进行常规处理
Sink任务向JobManager确认收到Checkpoints barrier,在所有任务成功将自身状态存入Checkpoints后整个应用的Checkpoints才算完成。
总结一下Checkpoint
和Savepoint
的区别和联系:
假设一种场景,从Kafka Source拉取数据,经过一次窗口聚合,最后将数据发送到Kafka Sink,如下图:
以上便是两阶段的完整流程,提交过程中如果失败有以下几种情况:
pre-commit
失败,将恢复到最近一次CheckPoint位置pre-commit
完成,必须要确保commit也要成功因此,所有opeartor必须对checkpoint最终结果达成共识:即所有operator都必须认定数据提交要么成功执行,要么被终止然后回滚。