当前位置：首页 > article >正文

大数据-127 - Flink State 04篇状态原理和原理剖析：状态存储 Part2

article 2025/1/11 15:11:17

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（已更完）
Flink（正在更新！）

章节内容

上节我们完成了如下的内容：

Flink 状态存储
MemoryStateBackend
FsStateBackend
RocksDBStateBackend
KeyedState
Operator State

在这里插入图片描述

上节进度

上节我们到了：
使用ManageOperatorState
（这里以及后续放到下一篇：大数据-127 Flink）

接下来我们继续上节的内容

使用ManageOperatorState

我们可以通过实现CheckpointedFunction或ListCheckpointed接口来使用 ManagedOperatorState。
CheckpointFunction
CheckpointFunction接口提供了访问 non-keyed state的方法，需要实现如下两个方法：

void snapshotState(FunctionSnapshotContext context) throws Exception;
void initializeState(FunctionInitializationContext context) throws Exception;

进行Checkpoint时会调用snapshotState()，用户自定义函数化时会调用 initializeState()，初始化包括第一次自定义函数初始化和从之前的Checkpoint恢复。因此 initializeState()，不仅是定义不同的状态类型初始化的地方，也需要包括状态恢复的逻辑。
当前，ManagedOperatorState以list的形式存在，这些状态是一个可序列化对象的集合List，彼此独立，方便在改变后进行状态的重新分派，换句话说，这些对象是重新分配non-keyed state的最新粒度，根据状态不同访问方式，有如下几种重新分配的模式：

Event-split redistribution：每个算子都保存一个列表形式的状态集合，整个状态由所有的列表拼接而成，当作业恢复或重新分配的时候，整个状态会按照算子的并发度进行均匀分配。比如说，算子A的并发读为1，包含两个元素element1和element2，当并发增加为2时，element1会被分发到并发0上，element2会被分发到并发1上。
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则需要处理恢复回来的对象列表，如果状态不可切分，则可以在snapshotState()中返回，Collections.singletonList(MY_STATE)。

StateBackend 如何保存

上面我们介绍了三种 StateBackend：

MemoryStateBackend
FsStateBackend
RocksDBStateBackend

在这里插入图片描述
在Flink的实际实现中，对于同一种StateBackend，不同的State在运行时会有细分的StateBackend托管，例如：MemoryStateBackend，就有DefaultOperatorStateBackend管理OperatorState，HeapKeyedStateBackend管理KeyedState。

我们看到MemoryStateBackend和FsStateBackend对于KeyedState和OperatorState的存储都符合我们之前的理解，运行时State数据保存于内存，checkpoint的保存位置需要注意下，并不是在RocksDB中，而是通过DefaultOperatorStateBackend保存于TaskManager内存。创建的源码如下：

// RocksDBStateBackend.java
// 创建 keyed statebackend
public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(...){
    ...
    return new RocksDBKeyedStateBackend<>(
    ...);
}
// 创建 Operator statebackend
public OperatorStateBackend createOperatorStateBackend(
    Environment env, String operatorIdentifier) throws Exception {
        //the default for RocksDB; eventually there can be a operator state
        backend based on RocksDB, too.
        final boolean asyncSnapshots = true;
        return new DefaultOperatorStateBackend(
    ...);
}

源码中也标注了，未来会提供基于RocksDB存储的OperatorState，所以当前即使使用RocksDBStateBackend，OperatorState也会超过内存限制。

Operator State 在内存中对应两种数据结构：
数据结构1：ListState 对应的实际实现类为 PartitionableListState，创建并注册的代码如下：

// DefaultOperatorStateBackend.java
private <S> ListState<S> getListState(...){
    partitionableListState = new PartitionableListState<>(
        new RegisteredOperatorStateBackendMetaInfo<>(
            name,
            partitionStateSerializer,
            mode));
    registeredOperatorStates.put(name, partitionableListState);
}

PartitionableListState中通过ArrayList来保存State数据：

// PartitionableListState.java
/**
* The internal list the holds the elements of the state
*/
private final ArrayList<S> internalList;

数据结构2：BroadcastState 对应的实际实现类为 HeapBroadcastState
创建并注册的代码如下：

public <K, V> BroadcastState<K, V> getBroadcastState(...) {
    broadcastState = new HeapBroadcastState<>(
        new RegisteredBroadcastStateBackendMetaInfo<>(
            name,
            OperatorStateHandle.Mode.BROADCAST,
            broadcastStateKeySerializer,
            broadcastStateValueSerializer));
    registeredBroadcastStates.put(name, broadcastState);
}

HeapBroadcastState中通过HashMap来保存State数据：

/**
* The internal map the holds the elements of the state.
*/
private final Map<K, V> backingMap;
HeapBroadcastState(RegisteredBroadcastStateBackendMetaInfo<K, V> stateMetaInfo) {
    this(stateMetaInfo, new HashMap<>());
}

配置StateBackend

我们知道Flink提供了三个StateBackend，那么如何配置使用某个StateBackend呢？默认的配置在conf/flink-conf.yaml文件中 state.backend 指定，如果没有配置该值，就会使用 MemoryStateBackend，默认的是StateBackend可以被代码中的配置覆盖。

Per-job设置

我们可以通过StreamExecutionEnvironment设置：

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

如果想使用RocksDBStateBackend，你需要将相关依赖加入你的Flink中：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
  <version>${flink.version}</version>
  <scope>provided</scope>
</dependency>

默认设置

如果没有在程序中指定，Flink将使用 conf/flink-conf.yaml文件中的 state.backend 指定的 StateBackend，这个值有三种配置：

JobManager（代表 MemoryStateBackend）
FileSystem（代表FsStateBackend）
RocksDB（代表RocksDBStateBackend）

开启Checkpoint

开启CheckPoint后，StateBackend管理的TaskManager上的状态数据才会被定期备份到JobManager或外部存储，这些状态数据在作业失败恢复时会用到。我们可以通过以下代码开启和配置CheckPoint：

StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
//env.getConfig().disableSysoutLogging();
//每 30 秒触发一次 checkpoint，checkpoint 时间应该远小于（该值 + MinPauseBetweenCheckpoints），否则程序会一直做checkpoint，影响数据处理速度
env.enableCheckpointing(30000); // create a checkpoint every 30 seconds
// set mode to exactly-once (this is the default)
// flink 框架内保证 EXACTLY_ONCE
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// make sure 30 s of progress happen between checkpoints
// 两个 checkpoints之间最少有 30s 间隔（上一个checkpoint完成到下一个checkpoint开始，默认 为0，这里建议设置为非0值）
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
// checkpoints have to complete within one minute, or are discarded
// checkpoint 超时时间（默认 600 s）
env.getCheckpointConfig().setCheckpointTimeout(600000);
// allow only one checkpoint to be in progress at the same time
// 同时只有一个checkpoint运行（默认）
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// enable externalized checkpoints which are retained after job cancellation
// 取消作业时是否保留 checkpoint (默认不保留)
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
// checkpoint失败时 task 是否失败( 默认 true， checkpoint失败时，task会失败)
env.getCheckpointConfig().setFailOnCheckpointingErrors(true);
// 对 FsStateBackend 刷出去的文件进行文件压缩，减小 checkpoint 体积
env.getConfig().setUseSnapshotCompression(true);

FsStateBackend 和 RocksDBStateBackend CheckPoint完成后最终保存到下面的目录：

hdfs:///your/checkpoint/path/{JOB_ID}/chk-{CHECKPOINT_ID}/

JOB_ID是应用的唯一ID，CHECK_POINT_ID 是每次 CheckPoint时自增的数字ID，我们可以从备份的CheckPoint数据恢复当时的作业状态。

flink-1x.x/bin/flink run -s hdfs:///your/checkpoint/path/{JOB_ID}/chk-{CHECKPOINT_ID}/ path/to//your/jar

我们可以实现 CheckpointedFunction 方法，在程序初始化的时候修改状态：

public class StatefulProcess extends KeyedProcessFunction<String, KeyValue, KeyValue> implements CheckpointedFunction {
    ValueState<Integer> processedInt;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }
    @Override
    public void processElement(KeyValue keyValue, Context context,
        Collector<KeyValue> collector) throws Exception {
        try{
            Integer a = Integer.parseInt(keyValue.getValue());
            processedInt.update(a);
            collector.collect(keyValue);
        }catch(Exception e){
            e.printStackTrace();
        }
    }
    @Override
    public void initializeState(FunctionInitializationContext
        functionInitializationContext) throws Exception {
        processedInt = functionInitializationContext.getKeyedStateStore().getState(new ValueStateDescriptor<>("processedInt", Integer.class));
        if(functionInitializationContext.isRestored()){
            //Apply logic to restore the data
        }
    }
    @Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        processedInt.clear();
    }
}
```