Flink 监控状态案例之温度跳变发警告

前言
【Flink 监控状态案例之温度跳变发警告】在项目中，经常遇到这样的场景，对于一批源源不断进入flink的数据源，需要检测某种类型的数据连续两次之间的数值变化范围，如果这个变化的值大于或者小于一定的标准值，将给出相应的告警；
在上一篇关于flink的常用状态管理的总结文章中，我们了解到了flink的常用的几种状态，如果应对这个场景，该使用哪种状态管理比较好呢？很明显是键控状态了；
通常来说，在实际的业务数据流中，都会有一些唯一标识数据的字段，那么通过这个字段做keyby的操作，接下来就可以使用键控状态做处理了；
下面看具体的代码实现：
import com.congge.source.SensorReading;import org.apache.flink.api.common.functions.RichFlatMapFunction;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 KeyedStateApplicationCase1 {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);// socket文本流DataStream inputStream = env.socketTextStream("localhost", 7777);// 转换成SensorReading类型DataStream dataStream = inputStream.map(line -> {String[] fields = line.split(",");return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));});// 定义一个flatmap操作，检测温度跳变，输出报警SingleOutputStreamOperator resultStream = dataStream.keyBy("id").flatMap(new TempChangeWarning(10.0));resultStream.print();env.execute();}public static class TempChangeWarning extends RichFlatMapFunction>{// 私有属性，温度跳变阈值private Double threshold;public TempChangeWarning(Double threshold) {this.threshold = threshold;}// 定义状态，保存上一次的温度值private ValueState lastTempState;@Overridepublic void open(Configuration parameters) throws Exception {lastTempState = getRuntimeContext().getState(new ValueStateDescriptor("last-temp", Double.class));}@Overridepublic void flatMap(SensorReading value, Collector 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());}@Overridepublic void close() throws Exception {lastTempState.clear();}}}
需要重点说明一点的是，这里keyby之后使用的是flatmap，因为flatmap会将过来的数据进行扁平化处理，同时由于需要记录数据的上下文状态，使用了RichFlatMapFunction；