Flink-状态管理-有状态算子和应用程序

迈不过友情╰ 2022-09-09 12:48 171阅读 0赞

# 状态管理 #

**流式计算分为无状态和有状态两种情况。**

无状态的计算观察每个独立事件,并根据最后一个事件输出结果。例如,流处理应用程序从传感器接收温度读数,并在温度超过 90 度时发出警告。有状态的计算则会基于多个事件输出结果。以下是一些例子。

所有类型的窗口。例如,计算过去一小时的平均温度,就是有状态的计算。

所有用于复杂事件处理的状态机。例如,若在一分钟内收到两个相差 20 度以上的温度读数,则发出警告,这是有状态的计算。

流与流之间的所有关联操作,以及流与静态表或动态表之间的关联操作,都是有状态的计算。

下图展示了无状态流处理和有状态流处理的主要区别。无状态流处理分别接收每条数据记录(图中的黑条),然后根据最新输入的数据生成输出数据(白条)。有状态流处理会维护状态(根据每条输入记录进行更新 ),并基于最新输入的记录和当前的状态值生成输出记录(灰条)。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]

上图中输入数据由黑条表示。无状态流处理每次只转换一条输入记录,并且仅根据最新的输入记录输出结果(白条)。有状态 流处理维护所有已处理记录的状态值,并根据每条新输入的记录更新状态,因此输出记录(灰条)反映的是综合考虑多个事件之后的结果。

尽管无状态的计算很重要,但是流处理对有状态的计算更感兴趣。事实上,正确地实现有状态的计算比实现无状态的计算难得多。旧的流处理系统并不支持有状态的计算,而新一代的流处理系统则将状态及其正确性视为重中之重。

![在这里插入图片描述][bec07bfa6592488099fc3dfd9bb3bca2.png]

由一个任务维护,并且用来计算某个结果的所有数据,都属于这个任务的状态；

可以认为状态就是一个本地变量，放在内存中，可以被任务的业务逻辑访问；

Flink 会进行状态管理,包括状态一致性、故障处理以及高效存储和访问,以便开发人员可以专注于应用程序的逻辑

## 有状态的算子和应用程序 ##

Flink 内置的很多算子,数据源 source,数据存储 sink 都是有状态的,流中的数据都是 buffer records,会保存一定的元素或者元数据。例如 : ProcessWindowFunction会缓存输入流的数据,ProcessFunction 会保存设置的定时器信息等等。

在 Flink 中,状态始终与特定算子相关联。 为了使运行时的 Flink 了解算子的状态,算子需要预先注册其状态。总的来说,有两种类型的状态:

**算子状态(operator state)**

**键控状态(keyed state)**

### 算子状态(operator state) ###

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

Flink 为算子状态提供三种基本数据结构:

列表状态(List state)：将状态表示为一组数据的列表。

联合列表状态(Union list state)：也将状态表示为数据的列表。它与常规列表状态的区别在于,在发生故障时,或者从保存点(savepoint)启动应用程序时如何恢复。

广播状态(Broadcast state)：如果一个算子有多项任务,而它的每项任务状态又都相同,那么这种特殊情况最适合应用广播状态。

算子状态只是针对当前分区的任务的状态进行维护和访问，所以与Key是没有关系的。在实际过程中大多数的情况还是需要与key有关系的，那么就需要用到键控状态了。

### 键控状态(keyed state) ###

键控状态是根据输入数据流中定义的键(key)来维护和访问的，只有当前key对应的数据才能访问当前key对应的状态。

Flink 为每个键值维护一个状态实例,并将具有相同键的所有数据,都分区到同一个算子任务中,这个任务会维护和处理这个 key 对应的状态。

当任务处理一条数据时,它会自动将状态的访问范围限定为当前数据的 key。

因此,具有相同 key 的所有数据都会访问相同的状态。Keyed State 很类似于一个分布式的 key-value map 数据结构,只能用于 KeyedStream (keyBy 算子处理之后)。

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_12_color_FFFFFF_t_70_g_se_x_16 1]

Flink 的 Keyed State 支持以下数据类型:

\*\*值状态ValueState\*\*保存单个的值,值的类型为 T。将状态表示为单个的值

get 操作: ValueState.value()

set 操作: ValueState.update(T value)

\*\*列表状态ListState\*\*保存一个列表,列表里的元素的数据类型为 T。将状态表示为一组数据的列表

ListState.add(T value)

ListState.addAll(List values)

ListState.get()返回 Iterable

ListState.update(List values)

\*\*映射状态MapState<K, V>\*\*保存 Key-Value 对。将状态表示为一组 Key-Value 对

MapState.get(UK key)

MapState.put(UK key, UV value)

MapState.contains(UK key)

MapState.remove(UK key)

\*\*聚合状态ReducingState和AggregatingState<I, O>\*\*将状态表示为一个用于聚合操作的列表

State.clear()是清空操作。

基本使用：

我们可以利用 Keyed State,实现这样一个需求:检测传感器的温度值,如果连续的两个温度差值超过 10 度,就输出报警

public static void main(String[] args) throws Exception { 
    
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
    
        // 从Socket读取
        DataStream<String> inputStream = env.socketTextStream("localhost", 4444);
    
        DataStream<SensorReading> dataStream = inputStream.map(line -> { 
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });
    
        // 定义一个flatMap操作，检查温度跳变，输出报警
        dataStream.keyBy("id")
            .flatMap(new TempChagneWarning(10.0)).print("temp-change");
    
    
        // 执行
        env.execute();
    
    }
    
    // 实现自定义函数类
    public static class TempChagneWarning
        extends RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>> { 
    
        // 私有属性，温度跳边阈值
        private Double threshold;
    
        public TempChagneWarning(Double threshold) { 
            this.threshold = threshold;
        }
    
        // 定义状态，保存上一次温度值
        private ValueState<Double> lastTemp;
    
        @Override
        public void open(Configuration parameters) throws Exception { 
            lastTemp = getRuntimeContext()
                .getState(new ValueStateDescriptor<Double>("last-temp", Double.class));
        }
    
        @Override
        public void flatMap(SensorReading value, Collector<Tuple3<String, Double, Double>> out) throws Exception { 
    
            // 获取状态
            Double lastTempData = lastTemp.value();
    
            // 如果状态不为null，判断差值
            if (lastTempData != null) { 
                double diff = Math.abs(value.getTemperature() - lastTempData);
                if (diff >= threshold) { 
                    out.collect(new Tuple3<>(value.getId(), lastTempData, value.getTemperature()));
                }
            }
    
            // 更新状态
            lastTemp.update(value.getTemperature());
        }
    
        @Override
        public void close() throws Exception { 
            lastTemp.clear();
        }
    }
    
    lxj@lxj:~/workspace/Olt-Test/trunk/onos$ cd
    lxj@lxj:~$ nc -lk 4444
    sensor_1,1547718209,31
    sensor_1,1547718209,35
    sensor_1,1547718209,49
    sensor_6,1547718201,15.4
    sensor_1,1547718209,30
    sensor_6,1547718201,50 
    
    temp-change> (sensor_1,35.0,49.0)
    temp-change> (sensor_1,49.0,30.0)
    temp-change> (sensor_6,15.4,50.0)

通过 RuntimeContext 注册 StateDescriptor。StateDescriptor 以状态 state 的名字和存储的数据类型为参数。

### 状态后端(State Backends) ###

每传入一条数据,有状态的算子任务都会读取和更新状态

由于有效的状态访问对于处理数据的低延迟至关重要,因此每个并行任务都会在本地维护其状态,以确保快速的状态访问

状态的存储、访问以及维护,由一个可插入的组件决定,这个组件就叫做**状态后端(state backend)**

状态后端主要负责两件事:本地的状态管理,以及将检查点(checkpoint)状态写入远程存储

**选择一个状态后端**

➢ **MemoryStateBackend**

内存级的状态后端,会将键控状态作为内存中的对象进行管理,将它们存储在TaskManager 的 JVM 堆上,而将 checkpoint 存储在 JobManager 的内存中

特点:快速、低延迟,但不稳定

一般不用于生产环境，多用于测试。

➢ **FsStateBackend**

将 checkpoint 存到远程的持久化文件系统(FileSystem)上,而对于本地状态,跟 MemoryStateBackend 一样,也会存在 TaskManager 的 JVM 堆上

同时拥有内存级的本地访问速度,和更好的容错保证。但是状态数据特别大的时候，会出现OOM的情况。

➢ **RocksDBStateBackend**

将所有状态序列化后,存入本地的 RocksDB 中存储。涉及到磁盘交互的话读写速度可能不高，但是不会出现OOM的情况。

注意:RocksDB 的支持并不直接包含在 flink 中,需要引入依赖:

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

设置状态后端为 FsStateBackend,并配置检查点和重启策略 :

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    
    // 1. 状态后端配置
    // env.setStateBackend(new MemoryStateBackend());
    env.setStateBackend(new FsStateBackend
    // env.setStateBackend(new RocksDBStateBackend(""));
    
    // 2. 检查点配置
    env.enableCheckpointing(1000);
    env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    env.getCheckpointConfig().setCheckpointTimeout(60000);
    env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
    env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
    env.getCheckpointConfig().setPreferCheckpointForRecovery(false);
    env.getCheckpointConfig().setTolerableCheckpointFailureNumber(0);
    
    // 3. 重启策略配置
    // 固定延迟重启(隔一段时间尝试重启一次)
    env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
        3, // 尝试重启次数
        100000 // 尝试重启的时间间隔,也可 org.apache.flink.api.common.time.Time
    ));

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/558c414cdc1742d192588bdcc1faaad2.png
[bec07bfa6592488099fc3dfd9bb3bca2.png]: /images/20220829/64ee8baf211d4c2fa1325d5620b5ce6e.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_12_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/39da8c81cd9140548cfc7ac6068f77e1.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6a2U56ybTG92ZQ_size_12_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/a4b84583c7b444eb814e341cccb521af.png