java Flink（十二）Flink状态之键控状态（keyed state）

Dear 丶 2022-11-16 13:47 313阅读 0赞

## **键控状态** ##

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

## **代码示例：** ##

**从kafka消费数据，解析连续两次温度，相差大于10就发生报警**

public static void main(String[] args) throws Exception {
            //获取当前执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //设置并行度
            env.setParallelism(1);
            //设置时间语义为时间时间
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            //从文件读取数据 ***demo模拟的文本流处理，其实不该用文本流处理，因为读取文本根本不许要分桶，速度过快了
            //DataStream<String> inputStream = env.readTextFile("D:\\idle\\FlinkTest\\src\\main\\resources");
            //从kafka读取数据
            DataStream<String> inputStream = env.addSource(new FlinkKafkaConsumer011<String>(kafkaConsumerTopic,
                    new SimpleStringSchema(), KafkaConsumerPro.getKafkaConsumerPro()));
    
            //转换成SensorReading类型
            DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<String, SensorReading>() {
                public SensorReading map(String line) throws Exception {
                    String[] fields = line.split(",");
                    return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
                }
            });
    
            //定义一个有状态的map操作. 基于key，先做keyBy
            SingleOutputStreamOperator<Tuple3<String, Double, Double>> resultStream = dataStream
                    .keyBy("id")
                    .flatMap(new MyKeyCountMapper(10.0));
        }
    
        //自定义实现richFunction 因为rich才可以访问上下文
        public static class MyKeyCountMapper extends RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>> {
            //定义一个value state,从上下文获取注册StateDescriptor。StateDescriptor 以状态 state 的名字和存储的数据类型为参数。
            //private ValueState<Integer> keyCountState;
            //其它类型状态声明
            //private ListState<String> listState;
    
            //私有属性，保存温度跳变阈值
            private Double threshold;
    
            public MyKeyCountMapper(Double threshold) {
                this.threshold = threshold;
            }
    
            //定义状态，保存上次温度值
            private ValueState<Double> lastTempState;
    
            @Override
            public void open(Configuration parameters) throws Exception {
                lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp", Double.class));
            }
    
            public void flatMap(SensorReading sensorReading, Collector<Tuple3<String, Double, Double>> collector) throws Exception {
                //获取上次温度值
                Double lastTemp = lastTempState.value();
                //如果不为null，那么判断两次温度差值
                if (lastTemp != null) {
                    Double diff = Math.abs(sensorReading.getTemperature() - lastTemp); //计算本次与上次温度差值
                    if (diff >= 10) {
                        collector.collect(new Tuple3<String, Double, Double>(sensorReading.getId(), lastTemp, sensorReading.getTemperature()));
                    }
                }
                //更新状态
                lastTempState.update(sensorReading.getTemperature());
            }
            @Override
            public void close() throws Exception {
                lastTempState.clear();
            }
        }