4.1.22 Flink-流处理框架-Flink中的状态管理之状态编程-温度跳变报警

Bertha 。 2022-11-05 08:32 135阅读 0赞

**目录**

1.需求

2.解决思路

3.代码

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

# 1.需求 #

利用 Keyed State，实现这样一个需求：检测传感器的温度值，如果连续的两个温度差值超过 10 度，就输出报警。

# 2.解决思路 #

我们需要定义一个flatmap操作，存储和更新上一次温度，与这一次温度进行比较，检测温度跳变是否超过阈值，输出报警

# 3.代码 #

package com.atguigu.apitest.state;
    import com.atguigu.apitest.beans.SensorReading;
    import org.apache.flink.api.common.functions.FlatMapFunction;
    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.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
    
    public class StateTest3_KeyedStateApplicationCase {
        public static void main(String[] args) throws Exception{
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(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]));
            });
    
            // 定义一个flatmap操作，检测温度跳变，输出报警
            SingleOutputStreamOperator<Tuple3<String, Double, Double>> resultStream = dataStream.keyBy("id")
                    .flatMap(new TempChangeWarning(10.0));
    
            resultStream.print();
    
            env.execute();
        }
    
        // 实现自定义函数类
        public static class TempChangeWarning extends RichFlatMapFunction<SensorReading, Tuple3<String, Double, Double>>{
            // 私有属性，温度跳变阈值
            private Double threshold;
    
            public TempChangeWarning(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));
            }
    
            @Override
            public void flatMap(SensorReading value, Collector<Tuple3<String, Double, Double>> out) throws Exception {
                // 获取状态
                Double lastTemp = lastTempState.value();
    
                // 如果状态不为null，那么就判断两次温度差值
                if( lastTemp != null ){
                    Double diff = Math.abs( value.getTemperature() - lastTemp );
                    if( diff >= threshold )
                        out.collect(new Tuple3<>(value.getId(), lastTemp, value.getTemperature()));
                }
    
                // 更新状态
                lastTempState.update(value.getTemperature());
            }
    
            @Override
            public void close() throws Exception {
                lastTempState.clear();
            }
        }
    }