Flink中的状态管理

小灰灰 2023-01-17 03:28 170阅读 0赞

## 一、状态定义 ##

*  由一个任务维护，并且用来计算某个结果的所有数据，都属于这个任务的状态
 *  可以认为状态就是一个本地变量，可以被任务的业务逻辑访问
 *  Flink会进行状态管理，包括状态一致性、故障处理以及高效存储和访问，以便开发人员可以专注于应用程序的逻辑  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]

## 二、状态管理 ##

在Flink中，状态始终与特定算子相关联。为了使运行时的Flink了解算子的状态，算子需要预先注册其状态

总的来说，有两种类型的状态：

*  算子状态（Operator State）：算子状态的作用范围限定为算子任务
 *  键控状态（Keyed State）：根据输入数据流中定义的键（key）来维护和访问

### 2.1 算子状态（Operator State） ###

算子状态的作用范围限定为**算子任务**，由同一并行任务所处理的所有数据都可以访问到相同的状态；状态对于**同一子任务**而言是**共享的**；算子状态不能被相同或不同算子的另一个子任务访问  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]

#### 2.1.1 算子状态数据结构 ####

*  列表状态（List state）：将状态表示为一组数据的列表
 *  联合列表状态（Union list state）：也将状态表示为数据的列表。它与常规列表状态的区别在于，在发生故障时，或者从保存点（savepoint）启动应用程序时如何恢复
 *  广播状态（Broadcast state）：如果一个算子有多项任务，而它的每项任务状态又都相同，那么这种特殊情况最适合应用广播状态

#### 2.1.2 示例代码 ####

public class StateTest1_OperatorState { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            //定义一个有状态的map操作，统计当前分区数据个数
            SingleOutputStreamOperator<Integer> resultStream = dataStream.map(new MyCountMapper());
            resultStream.print();
    
            env.execute();
        }
    
        //自定义MapFunction
        public static class MyCountMapper implements MapFunction<SensorReading, Integer>, ListCheckpointed<Integer> { 
    
            //定义一个本地变量，作为算子状态
            private Integer count = 0;
    
            @Override
            public Integer map(SensorReading value) throws Exception { 
                count++;
                return count;
            }
    
            @Override
            public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception { 
                return Collections.singletonList(count);
            }
    
            @Override
            public void restoreState(List<Integer> state) throws Exception { 
                for(Integer num : state){ 
                    count += num;
                }
            }
        }
    }

### 2.2 键控状态（Keyed State） ###

键控状态是根据输入数据流中**定义的键（key）来维护和访问的；Flink为每个key**维护**一个状态实例**，并将具有相同键的所有数据，都**分区到同一个算子任务**中，这个任务会维护和处理这个key对应的状态；当任务处理一条数据时，它会自动将状态的访问范围限定为当前数据的key

#### 2.2.1 键控状态数据结构 ####

*  值状态（Value state）：将状态表示为单个的值
 *  列表状态（List state）：将状态表示为一组数据的列表
 *  映射状态（Map state）：将状态表示为一组 Key-Value 对
 *  聚合状态（Reducing state & Aggregating state）：将状态表示为一个用于聚合操作的列表

#### 2.2.2 键控状态的使用 ####

*  声明一个键控状态

myValueState = getRuntimeContext().getState(
    	new ValueStateDescriptor<Integer>(
    		"my-value",
    		Integer.class));

*  读取状态

Integer myValue = myValueState.value();

*  对状态赋值

myValueState.update(10);

**值状态的示例代码:**

**其它状态的示例代码：**

public class StateTest2_KeyedState { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            //定义一个有状态的map操作，统计当前sensor数据个数
            SingleOutputStreamOperator<Integer> resultStream = dataStream.keyBy("id")
                    .map(new MyKeyCountMapper());
            resultStream.print();
            env.execute();
        }
    
        //自定义实现一个RichMapFunction
        public static class MyKeyCountMapper extends RichMapFunction<SensorReading, Integer>{ 
            //声明一个键控状态
            private ValueState<Integer> keyCountState;
    
            //其它类型状态的声明
            private ListState<String> myListState;
            private MapState<String, Double> myMapState;
            private ReducingState<SensorReading> myReducingState;
    
            @Override
            public void open(Configuration parameters) throws Exception { 
                keyCountState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("key-count", Integer.class));
    
                myListState = getRuntimeContext().getListState(new ListStateDescriptor<String>("my-list", String.class));
    
                myMapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, Double>("my-map", String.class, Double.class));
    
    //            myReducingState = getRuntimeContext().getReducingState(new ReducingStateDescriptor<SensorReading>("my-reducing", , ));
            }
    
            @Override
            public Integer map(SensorReading value) throws Exception { 
                //所有状态API都有clear()清空操作
    
                //其它状态API调用
                //list state
                for(String str : myListState.get()){ 
                    System.out.println(str);
                }
                myListState.add("hello");
    
                //map state
                myMapState.get("key1");
                myMapState.put("key2", 35.001);
                myMapState.remove("key2");
                myMapState.clear();
    
      **加粗样式**          //reducing state 聚合操作
                myReducingState.add(value);
                myReducingState.clear();
    
                //读取状态
                Integer count = keyCountState.value();
                count++;
                //对状态赋值
                keyCountState.update(count);
                return count;
            }
        }
    }

## 三、状态后端 ##

*  每传入一条数据，有状态的算子任务都会读取和更新状态
 *  由于有效的状态访问对于处理数据的低延迟至关重要，因此每个并行任务都会在**本地**维护其状态，以确保快速的状态访问
 *  状态的**存储、访问及维护**，由一个可插入的组件决定，这个组件就叫做状态后端（state backend）
 *  状态后端主要负责两件事：**本地的状态管理**、**以及检查点（checkpoint）状态写入远程存储**

### 3.1 状态后端类型 ###

*  MemoryStateBackend  
    **内存级**的状态后端，会将**键控状态**作为内存中的对象进行管理，将它们存储在TaskManager的JVM堆上，而将checkpoint存储在JobManager的内存中  
    **特点：** 快速、低延迟、但不稳定
 *  FsStateBackend  
    将checkpoint存到远程的持久化文件系统（FileSystem）上，而对于本地状态，跟MemoryStateBackend一样，也会存在TaskManager的JVM堆上  
    **特点：** 同时拥有内存级的本地访问速度，和更好的容错保证
 *  RocksDBStateBackend（适用于管理的**状态多且对实时性要求不高**的场景）  
    将所有状态序列化后，存入本地的RocksDB中

### 3.2 状态后端的配置 ###

**方式一：**  
可以在flink安装包下找到 conf/flink-conf.yaml 文件，修改相应的状态后端配置

**方式二：**  
可以在代码中进行配置

pom.xml

<dependency>
         <groupId>org.apache.flink</groupId>
         <artifactId>flink-statebackend-rocksdb_2.12</artifactId>
         <version>1.10.1</version>
    </dependency>

代码

public class StateTest3_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(""));
    
            // socket文本流
            DataStream<String> inputStream = env.socketTextStream("localhost", 7777);
    
            // 转换成SensorReading类型
            // java8 中的lamda表达式
            DataStream<SensorReading> dataStream = inputStream.map(line -> { 
                String[] fields = line.split(",");
                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
            });
    
            env.execute();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70]: /images/20221022/691ab15b642c4d5eb9bedb4500c42bf1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM0MjQxODYx_size_16_color_FFFFFF_t_70 1]: /images/20221022/34154c46287c4a7c988e27019236cb3c.png