巫山那家做网站,公司官网制作多少钱,短视频推广app,wordpress特效代码State Backends
由 Flink 管理的 keyed state 是一种分片的键/值存储#xff0c;每个 keyed state 的工作副本都保存在负责该键的 taskmanager 本地中。另外#xff0c;Operator state 也保存在机器节点本地。Flink 定期获取所有状态的快照#xff0c;并将这些快照复制到持…State Backends
由 Flink 管理的 keyed state 是一种分片的键/值存储每个 keyed state 的工作副本都保存在负责该键的 taskmanager 本地中。另外Operator state 也保存在机器节点本地。Flink 定期获取所有状态的快照并将这些快照复制到持久化的位置例如分布式文件系统。
如果发生故障Flink 可以恢复应用程序的完整状态并继续处理就如同没有出现过异常。
Flink 管理的状态存储在 state backend 中。Flink 有两种 state backend 的实现
一种基于 RocksDB 内嵌 key/value 存储将其工作状态保存在磁盘上的将其状态快照持久化到分布式文件系统另一种基于堆的 state backend将其工作状态保存在 Java 的堆内存中。这种基于堆的 state backend 有两种类型 FsStateBackend将其状态快照持久化到分布式文件系统MemoryStateBackend它使用 JobManager 的堆保存状态快照。 当使用基于堆的 state backend 保存状态时访问和更新涉及在堆上读写对象。但是对于保存在 RocksDBStateBackend 中的对象访问和更新涉及序列化和反序列化所以会有更大的开销。但 RocksDB 的状态量仅受本地磁盘大小的限制。还要注意只有 RocksDBStateBackend 能够进行增量快照这对于具有大量变化缓慢状态的应用程序来说是大有裨益的。
所有这些 state backends 都能够异步执行快照这意味着它们可以在不妨碍正在进行的流处理的情况下执行快照。
Checkpoint
Flink 定期对每个算子的所有状态进行持久化快照并将这些快照复制到更持久的地方例如分布式文件系统。 如果发生故障Flink 可以恢复应用程序的完整状态并恢复处理就好像没有出现任何问题一样。
这些快照的存储位置是通过作业_checkpoint storage_定义的。 有两种可用检查点存储实现一种持久保存其状态快照 到一个分布式文件系统另一种是使用 JobManager 的堆。 Flink不同版本StateBackend状态与Checkpoint Storage快照 关系
在Flink1.14之前StateBackend与Checkpoint Storage 耦合在一起但在Flink1.14之后把StateBackend与Checkpoint Storage 实现了解耦使逻辑更加清晰。
Flink1.14之前
基于 RocksDB state backend状态快照持久化到分布式文件系统
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new RocksDBStateBackend(hdfs://namenode:8020/data/rocksdb/ck, true));//true: 增量checkpoint; false:全量checkpoint
env.setStateBackend(new RocksDBStateBackend(file:///data/rocksdb/ck, true));//本地文件系统基于heap state backend状态快照持久化到分布式文件系统
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new FsStateBackend(hdfs://namenode:8020/data/fs/ck));//远程分布式文件系统
env.setStateBackend(new FsStateBackend(file:///data/fs/ck));//本地文件系统基于heap state backend使用 JobManager 的堆保存状态快照。
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new MemoryStateBackend());Flink1.14之后推荐使用
基于 RocksDB state backend状态快照持久化到分布式文件系统
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));//true: 增量checkpoint; false:全量checkpoint
env.getCheckpointConfig().setCheckpointStorage(hdfs://namenode:8020/data/rocksdb/ck);//远程分布式文件系统
env.getCheckpointConfig().setCheckpointStorage(file:///data/rocksdb/ck);//本地文件系统
flink-conf.yaml配置 state.backend: rocksdbstate.checkpoints.dir: hdfs:///checkpoints/基于heap state backend状态快照持久化到分布式文件系统
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage(hdfs://namenode:8020/data/fs/ck);//远程分布式文件系统
env.getCheckpointConfig().setCheckpointStorage(file:///data/fs/ck);//本地文件系统flink-conf.yaml配置 state.backend: hashmapstate.checkpoints.dir: hdfs:///checkpoints/基于heap state backend使用 JobManager 的堆保存状态快照。
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());flink-conf.yaml配置 state.backend: hashmapstate.checkpoint-storage: jobmanager总结
默认情况下 checkpoint 是禁用的需要手动开启 env.enableCheckpointing(long interval, CheckpointingMode mode) 默认情况下StateBackend是保持在 TaskManagers 的heap内存中checkpoint 保存在 JobManager 的内存中。只有基于 RocksDB state backend的状态快照才支持增量checkpoint基于heap的并不支持Flink状态分为Keyed State和非keyed State: Keyed State可以使用RocksDB state backend和heap state backend。 所有支持的状态类型如下所示 ValueState: 保存一个可以更新和检索的值如上所述每个值都对应到当前的输入数据的 key因此算子接收到的每个 key 都可能对应一个值。 这个值可以通过 update(T) 进行更新通过 T value() 进行检索。 ListState: 保存一个元素的列表。可以往这个列表中追加数据并在当前的列表上进行检索。可以通过 add(T) 或者 addAll(List) 进行添加元素通过 Iterable get() 获得整个列表。还可以通过 update(List) 覆盖当前的列表。 ReducingState: 保存一个单值表示添加到状态的所有值的聚合。接口与 ListState 类似但使用 add(T) 增加元素会使用提供的 ReduceFunction 进行聚合。 AggregatingStateIN, OUT: 保留一个单值表示添加到状态的所有值的聚合。和 ReducingState 相反的是, 聚合类型可能与 添加到状态的元素的类型不同。 接口与 ListState 类似但使用 add(IN) 添加的元素会用指定的 AggregateFunction 进行聚合。 MapStateUK, UV: 维护了一个映射列表。 你可以添加键值对到状态中也可以获得反映当前所有映射的迭代器。使用 put(UKUV) 或者 putAll(MapUKUV) 添加映射。 使用 get(UK) 检索特定 key。 使用 entries()keys() 和 values() 分别检索映射、键和值的可迭代视图。你还可以通过 isEmpty() 来判断是否包含任何键值对。 非keyed State不使用 RocksDB state backend需要保存在内存中包括 算子状态 (Operator State)广播状态 (Broadcast State)尤其需要考虑保证充足的内存自定义 Operator StateCheckpointedFunction 接口提供了访问 non-keyed state 的方法需要实现如下两个方法 void snapshotState(FunctionSnapshotContext context) throws Exception; void initializeState(FunctionInitializationContext context) throws Exception;
参考
Fault Tolerance via State Snapshots
