Flink——状态管理

状态分类：算子状态（Operatior State）；键控状态（Keyed State）；状态后端(StateBackends）

什么是状态

1：由一个任务维护，并且用来计算某个结果的所有数据，都属于这个任务的状态 2：可以认为状态就是一个本地变量，可以被任务的业务逻辑访问 3：Flink 会进行状态管理，包括状态一致性、故障处理以及高效存储和访问，以便开发人员可以专注于应用程序的逻辑 算子状态（Operator State）

1：算子状态的作用范围限定为算子任务，由同一并行任务所处理的所有数据都可以访问到相同的状态 2：状态对于同一子任务而言是共享的 3： 算子状态不能由相同或不同算子的另一个子任务访问

列表状态（List state）：将状态表示为一组数据的列表

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

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

class MyCountMap implements MapFunction>, ListCheckpointed {

    //定义算子状态
    private Integer count = 0;
    
    @Override
    public Tuple2 map(SensorReading value) throws Exception {
        count++;
        return new Tuple2<>(value.getId(),count);
    }

    @Override
    public List snapshotState(long checkpointId, long timestamp) throws Exception {
        return Collections.singletonList(count);
    }

    @Override
    public void restoreState(List state) throws Exception {
        for (Integer integer : state) {
            count+=integer;
        }
    }
}

键控状态（Keyed State） 1：键控状态是根据输入数据流中定义的键（key）来维护和访问的 2：Flink 为每个 key 维护一个状态实例，并将具有相同键的所有数据，都分区到 3：同一个算子任务中，这个任务会维护和处理这个 key 对应的状态 4：当任务处理一条数据时，它会自动将状态的访问范围限定为当前数据的 key 1：值状态（Value state） ：将状态表示为单个的值 2：列表状态（List state） ：将状态表示为一组数据的列表                                                            3：映射状态（Map state）将状态表示为一组 Key-Value 对 4：聚合状态（Reducing state & Aggregating State） ：将状态表示为一个用于聚合操作的列表

package com.atguigu.state;

import com.atguigu.bean.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import scala.Tuple2;

import java.util.Collections;
import java.util.List;

public class KeyedState {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource socketTextStream = env.socketTextStream("hadoop112", 7777);
        DataStream inputStream = socketTextStream.map(value -> {
            String[] split = value.split(",");
            return new SensorReading(split[0], new Long(split[1]), new Double(split[2]));
        });

        
        SingleOutputStreamOperator> map = inputStream.keyBy("id").map(new MyCountMap2());

        map.print();
        env.execute();
    }
}

class MyCountMap2 extends RichMapFunction> {

    //定义算子状态
    private ValueState myKed ;

    @Override
    public void open(Configuration parameters) throws Exception {
        myKed = getRuntimeContext().getState(new ValueStateDescriptor("key_count",Integer.class,0));
    }

    @Override
    public Tuple2 map(SensorReading value) throws Exception {
        Integer count = myKed.value();
        count++;
        myKed.update(count);
        return new Tuple2<>(value.getId(),count);
    }
    
}

状态后端（State Backends） 1：每传入一条数据，有状态的算子任务都会读取和更新状态 2：由于有效的状态访问对于处理数据的低延迟至关重要，因此每个并行任务都会在本地维护其状态，以确保快速的状态访问 状态的存储、访问以及维护，由一个可插入的组件决定，这个组件就 叫做 状态后端 （state backend） 3：状态后端主要负责两件事：本地的状态管理，以及将检查点（checkpoint）状态写入远程存储 选择一个状态后端 ➢ MemoryStateBackend • 内存级的状态后端，会将键控状态作为内存中的对象进行管理，将它们存储在TaskManager 的 JVM 堆上，而将 checkpoint 存储在 JobManager 的内存中 • 特点：快速、低延迟，但不稳定 ➢ FsStateBackend • 将 checkpoint 存到远程的持久化文件系统（FileSystem）上，而对于本地状态，跟 MemoryStateBackend 一样，也会存在 TaskManager 的 JVM 堆上同时拥有内存级的本地访问速度，和更好的容错保证 ➢ RocksDBStateBackend • 将所有状态序列化后，存入本地的 RocksDB 中存储。 状态编程

同一个传感器前后温度相差10就报警输出：(sensor_1,35.8,50.0)

package com.atguigu.state;

import com.atguigu.bean.SensorReading;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import scala.Tuple2;

public class KeyedStateTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource socketTextStream = env.socketTextStream("hadoop112", 7777);
        DataStream inputStream = socketTextStream.map(value -> {
            String[] split = value.split(",");
            return new SensorReading(split[0], new Long(split[1]), new Double(split[2]));
        });

        
        SingleOutputStreamOperator> map = inputStream.keyBy("id").flatMap(new MyFlatMap(10.0));

        map.print();
        env.execute();
    }
}

class MyFlatMap extends RichFlatMapFunction> {

    //当前温度跳变差值
    private double temp;

    public MyFlatMap(double d){
        this.temp = d;
    }

    private ValueState lastTempState = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        lastTempState = getRuntimeContext().getState(new ValueStateDescriptor("last_State",Double.class));
    }

    @Override
    public void flatMap(SensorReading value, Collector> out) throws Exception {
       //获取上次的温度值
       Double  lastTemp = lastTempState.value();

       if(lastTemp!=null){
           Double diff = Math.abs(value.getTemperature()-lastTemp);
           if(diff>=temp){
               out.collect(new Tuple3<>(value.getId(),lastTemp,value.getTemperature()));
           }
       }

       //更新状态
       lastTempState.update(value.getTemperature());
    }

    @Override
    public void close() throws Exception {
        lastTempState.clear();
    }
}