「Flink」Flink的状态管理与容错
在Flink中的每个函数和运算符都是有状态的。在处理过程中可以用状态来存储数据,这样可以利用状态来构建复杂操作。为了让状态容错,Flink需要设置checkpoint状态。Flink程序是通过checkpoint来保证容错,通过checkpoint机制,Flink可恢复作业的状态和计算位置。
checkpoint检查点
前提条件
Flink的checkpoin机制需要与流和状态的持久化存储交互,一般它要求:
- 一个持久化的数据源
- 当Flink程序出现问题时,可以通过checkpoint持久化存储中恢复,然后从出错的地方开始重新消费数据
- 该数据源可以在一定时间内重跑数据,例如:Kafka、RabbitMQ或者文件系统HDFS、S3、…
- 状态的持久存储
- 状态需要永久的保存下来,通常是分布式文件系统(例如:HDFS、S3、GFS、…)
启用和配置检查点
默认情况,Flink是禁用检查点。要启用检查点,调用
// 启用检查点// 单位:毫秒
env.enableCheckpointing(1000);
在启用检查点时,还可以配置检查点的其他参数。
- exactly-one or at-least-once(仅一次或者至少一次)
- 大多数程序都是设置为exactly-once,只有在某些超低延迟的应用(例如:始终要求是毫秒级的应用)
- 通过查看源码,我们看到,Flink默认是 exactly-once
public static final CheckpointingMode DEFAULT_MODE = CheckpointingMode.EXACTLY_ONCE;
- 检查点超过规定的时间就会自动终止
- 检查点之间的最小时间
- 下一个检查点将在上一个检查点完成后5秒钟启动
- 检查点最小间隔时间不会受检查点间隔更容易配置
- 检查点的并发数目。默认情况一个检查点在运行时不会触发另一个检查点,这样可以确保Flink不会花太多时间在checkpoint上,并确保流可以有效进行。
- 可以设置多个重叠的checkpoint,这对容许有一定延迟,并希望较频繁的检查(100ms)来重新处理故障是有用的
- 外部检查点
- 可以将检查点设置为外部持久化,这样检查点的元数据将写入持久存储,并且但作业运行失败是不会自动清理
- 这样可以做双重保险
- 检查点执行发生错误,是否执行任务。
- 默认情况,如果checkpoint失败,任务也将失败
- 即时最近有更多的savepoint可用于恢复,flink依然会选择使用最近一次的checkpoint来进行错误恢复
参考配置:
// --------// 配置checkpoint
// 启用检查点
env.enableCheckpointing(1000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
选择状态后端存储
Flink的checkpoint机制可以存储计时器和有状态operation的所有快照,包括:连接器、窗口或者用户自定义状态。具体checkpoint存储在哪儿(例如:是JobManager内存、文件系统或者数据库),依赖于状态后端的配置。
默认情况,状态保存在TaskManager的内存中,检查点存储在TM的内存中。为了适当地保存大状态,Flink支持其他的存储。我们可以通过:
StreamExecutionEnvironment.setStateBackend(…)
来指定存储方式
Flink状态管理
状态的应用场景:
- 当应用程序想要按照某种模式搜索某些事件时,状态可以保存迄今所有的事件序列
- 当每分钟/小时/天需要对流数据进行聚合,状态可以保存挂起的聚合
- 当在数据流上训练机器学习模型时,状态可以用来保存某一类参数的版本
- 当需要管理历史数据时,状态允许访问过去历史数据
Flink状态可以保存在堆内、或者是堆外。Flink也可以管理应用程序的状态,必要时也可以溢出到磁盘,如果应用要保持非常大的状态,可以不修改程序逻辑情况下配置状态后端存储。
Flink状态分类
Flink中有两种基本的状态:
- Keyed State
- Operator State
Keyed State
Keyed State通常和key相关,仅仅在KeyedStream的方法和算子中使用。可以把 Keyed State看作是分区,而且每一个key仅出现在一个分区内。逻辑上每个 keyed-state和唯一元组<算子并发实例, key>绑定,由于每个key仅属于算子的一个并发,因此可以简化为<算子, key>
Operator State
对于 Operator State来说,每个Operator State和一个并发实例绑定。Kafka connector是Flink中使用operator state的一个很好的示例。每个Kafka消费者的并发在Operator State中维护一个 topic partition到offset的映射关系。
Operator state在Flink作业的并发改变后,会重新分发状态,分发的策略和keyed stated不一样。
Raw State与Managed State
Keyed Stated和Operator State分别有两种形式:managed 和 raw
Managed State是由Flink运行时管理的数据结构来表示的,例如:内部的Hash Table或者RocksDB。例如:ValueState、ListState等。Flink运行时会对这些状态进行编码并写入Checkpoint。
Raw State则保存在自己的数据结构中。checkpoint的时候,Flink并不知道状态里面具体的内容,仅仅写入一串字节序列到checkpoint中。
所有的DataStream的function都可以使用managed state,但raw state只能在实现算子时使用。由于Flink可以在修改并发时更好的分发状态数据,并且能够更好的管理内存,因为讲义使用 managed state.
使用Managed Keyed State
Managed keyed state接口提供不同类型的状态访问接口,这些状态都作用在当前输入数据的key下。这些状态仅可在KeyedStream上使用,可以通过 stream.keyBy(…)得到KeyedStream。
所有支持的状态类型如下:
- ValueState<T>
- 保存一个可以更新和获取的值,算子接收到的每个key都可能对应一个值
- 可以通过update(T)进行更新,通过value()获取
- ListState<T>
- 保存一个元素的列表,可以往这个列表中追加数据,并在当前列表上检索
- 可以通过 add(T)或者addAll(List<T>)进行追加元素
- 通过get()获取整个列表
- 通过 update(List<T>)覆盖当前列表
- ReducingState<T>
- 保存一个单值,表示添加到状态的所有值的聚合。接口与ListState类似
- AggregatingState<IN, OUT>
- 保存一个单值,表示添加到状态的所有值的聚合
- 与ReducingState相反的是,聚合类型可能与添加到状态的元素类型不同。接口与ListState类似
- FoldingState<T, ACC>(后续将过期)
- 保存一个单值,白搜狐添加到状态的所有值的集合
- 与ReducingState相反的是,聚合类型可能与添加到状态的元素类型不同。接口与ListState类似
- MapState<UK, UV>
- 维护一个映射列表,可以添加键值到状态中,可以获取当前映射的迭代器
- 使用put、putAll添加映射,使用 get检索特定key
注意:
- 这些状态对象仅用于状态交互。状态本身不一定存储在内存中,还有可能保存在磁盘或者其他位置
- 从状态中获取的值取决于输入元素说代表的key,因此,在不同key上调用同一个接口,可能得到不同的值
使用Managed Operator State
可以通过实现 CheckpointedFunction 或者 ListCheckpointed<T extends Serialized>接口来使用Managed Operator State。
CheckpointedFunction接口:
void snapshotState(FunctionSnapshotContext context) throws Exception;void initializeState(FunctionInitializationContext context) throws Exception;
在Flink进行checkpoint时,会调用snapshotstate(),用户自定义函数初始化时会调用 initializeState。初始化包括第一次自定义函数初始化和从之前的 checkpoint 回复。因此,initializeState 中应该也包括状态恢复的逻辑。
Managed Operator State以list的形式存在,这些状态是一个可序列化对象的集合List,彼此独立,方便在改变并发后进行状态的重新分派。换句话说,这些对象是重新分配 non-keyed state的最细粒度。根据状态的不同访问方式,有以下两种分配模式:
- Even-split redistribution
- 每个算子都存储一个列表形式的状态集合,整个状态由所有的列表拼接而成
- 但作业恢复或者重新分配时,整个状态按照算子的并行度均匀分配
- Union redistribution
- 每个算子保存一个列表形式的状态集合,整个状态由所有的列表拼接而成
- 但作业恢复或者重新分配时,每个算子都将获得所有的状态数据
ListCheckpointed接口:
ListCheckpointed接口是CheckpointedFunction接口的精简版,仅支持 even-split redistribution的list state
List<T> snapshotState(long checkpointId, long timestamp) throws Exception;void restoreState(List<T> state) throws Exception;
snapshotState()需要返回一个将写入到checkpoint的对象列表, restoreState则需要处理恢复回来的对象列表。
Flink官方文档:
https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/dev/stream/state/checkpointing.html
https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/ops/state/checkpoints.html
https://ci.apache.org/projects/flink/flink-docs-release-1.9/zh/dev/stream/state/state.html
以上是 「Flink」Flink的状态管理与容错 的全部内容, 来源链接: utcz.com/z/532144.html