最近在哔站学习武老师Flink的教程，受益匪浅。其中关于Flink的一些案例，非常有意思，故记录下来。分享给感兴趣的朋友。
本次案例的场景是：假设有n个传感器，其温度值被记录下并灌到Kafka的主题里。我们需要读取这些数据并分析之，在10s内如果该温度值持续上升，则抛出警告信息。
实现方式：因为要在“指定时间内”温度“持续上升”，需要获取上下文状态，且需包含一个定时器的功能去指定一段时间，因此考虑到使用ProcessFunction，而因为要对n个传感器进行分组处理，所以考虑使用KeyedProcessFunction。
具体代码如下：
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;import java.util.Properties;public class TempConRiseWarningDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);
// DataStreamSource<String> inputStream = env.socketTextStream("localhost", 9999);Properties properties = new Properties();properties.setProperty("bootstrap.servers", "vm01:9092");properties.setProperty("group.id", "consumer-group");properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");properties.setProperty("auto.offset.reset", "latest");DataStreamSource<String> inputStream = env.addSource(new FlinkKafkaConsumer<String>("test", new SimpleStringSchema(), properties));KeyedStream<Tuple2<String, Double>, String> keyedStream = inputStream.map(new MapFunction<String, Tuple2<String, Double>>() {@Overridepublic Tuple2<String, Double> map(String value) throws Exception {String[] fields = value.split(",");return new Tuple2<>(fields[0], new Double(fields[1]));}}).keyBy(tuple2 -> tuple2.f0);keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));keyedStream.process(new TempConRiseWarning(10)).print();env.execute("Run TempConRiseWarningDemo");}static class TempConRiseWarning extends KeyedProcessFunction<String, Tuple2<String, Double>, String> {private Integer interval;public TempConRiseWarning(Integer interval) {this.interval = interval;}// 定义状态，保存上一次的温度值以及定时器时间戳private ValueState<Double> lastTempState;private ValueState<Long> timerTsState;@Overridepublic void open(Configuration parameters) throws Exception {lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp-state", Double.class));timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-timestamp-state", Long.class));}@Overridepublic void processElement(Tuple2<String, Double> value, Context ctx, Collector<String> out) throws Exception {// 取出值Double lastTemp = lastTempState.value();Long timerTs = timerTsState.value();// 如果上次温度值为null，赋值为Double的最小值if (lastTemp == null) {lastTemp = Double.MIN_VALUE;}// 如果温度上升并且没有定时器，注册定时器，开始等待if (lastTemp < value.f1 && timerTs == null) {// 获取定时器时间戳long ts = ctx.timerService().currentProcessingTime() + interval * 1000L;// 注册定时器ctx.timerService().registerProcessingTimeTimer(ts);// 设置时间戳timerTsState.update(ts);} else if (lastTemp > value.f1 && timerTs != null) { // 如果温度下降，删除定时器// 删除定时器ctx.timerService().deleteProcessingTimeTimer(timerTs);// 清除时间戳timerTsState.clear();}// 更新温度值lastTempState.update(value.f1);}@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {// 输出警告out.collect("Warning! the temperature of " + ctx.getCurrentKey() + " has risen continuously in last " + interval + " seconds.");// 清除定时器时间戳timerTsState.clear();}@Overridepublic void close() throws Exception {lastTempState.clear();timerTsState.clear();}}
}