4.1.23 Flink-流处理框架-Flink中的状态管理之状态后端

向右看齐 2022-11-06 14:58 264阅读 0赞

**目录**

1.状态后端是什么？

2.状态后端的类型

3.配置状态后端

3.1 配置文件

3.2 代码中针对单独job，设置状态后端

--------------------

# 1.状态后端是什么？ #

每传入一条数据，有状态的算子任务都会读取和更新状态，由于有效的状态访问对于处理数据的低延迟至关重要，因此每个并行任务都会在本地维护其状态，以确保快速的状态访问。**状态的存储、访问以及维护，由一个可插入的组件决定，这个组件就叫做状态后端(state backend)。**

状态后端主要负责两件事：1）本地的状态管理；2）将检查点(checkpoint)状态写入远程存储

# 2.状态后端的类型 #

在flink中给我们提供了三种不同的状态后端。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70][]

# 3.配置状态后端 #

## 3.1 配置文件 ##

配置文件可以**配置整个集群的状态后端**：conf/flink-conf.yaml文件

和我们上面的名字不太一样，下面的名字表示checkpoint往哪里存储，默认写到文件系统持久化，一般直接保存在hdfs上面即可。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1][]

## 3.2 代码中针对单独job，设置状态后端 ##

package com.atguigu.apitest.state;
    
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.api.common.restartstrategy.RestartStrategies;
    import org.apache.flink.api.common.time.Time;
    import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
    import org.apache.flink.runtime.state.filesystem.FsStateBackend;
    import org.apache.flink.runtime.state.memory.MemoryStateBackend;
    import org.apache.flink.streaming.api.CheckpointingMode;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    public class StateTest4_FaultTolerance {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // 1. 状态后端配置
            //env.setStateBackend( new MemoryStateBackend());
            env.setStateBackend( new FsStateBackend(""));
            //env.setStateBackend( new RocksDBStateBackend(""));
    
            // 2. 检查点配置
            env.enableCheckpointing(300);
    
            // 高级选项
            env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
            env.getCheckpointConfig().setCheckpointTimeout(60000L);
            env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100L);
            env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
            env.getCheckpointConfig().setTolerableCheckpointFailureNumber(0);
    
            // 3. 重启策略配置
            // 固定延迟重启
            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10000L));
            // 失败率重启
            env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(10), Time.minutes(1)));
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            DataStream<SensorReading> dataStream = inputStream.map(line -> {
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            dataStream.print();
            env.execute();
        }
    }

如果想要引入最后一种状态后端方式，RocksDBStateBackend，需要引入第三方库，2.12是scala版本，1.10.1是flink版本

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70]: /images/20221023/3eeccc2dca744f56ae8bafaf81c9004e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 1]: /images/20221023/d2070692615447a9be8b1d75e0c4b7ae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1N1eWViaXViaXU_size_16_color_FFFFFF_t_70 2]: /images/20221023/277a3d6587834446aa456465836981fe.png