Flink ReducingState 实例

ReducingState介绍

• ReducingState是和ReduceFunction配合使用
• get() 获取状态的值
• add(IN value)方法添加一个元素，触发reduceFunction计算一次

需求：输出各设备10s内最大温度

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Random;

public class ReducingStateTest {
    public static void main(String[] args) throws Exception {
        // 需求：取10秒内最高的温度进行输出
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setAutoWatermarkInterval(100l);
        DataStreamSource> tuple3DataStreamSource = env.addSource(new SourceFunction>() {
            boolean flag = true;

            @Override
            public void run(SourceContext> ctx) throws Exception {
                String[] str = {"水阀1", "水阀2", "水阀3"};
                while (flag) {
                    int i = new Random().nextInt(3);
                    // 温度
                    int temperature = new Random().nextInt(100);
                    Thread.sleep(1000l);
                    // 设备号、温度、事件时间
                    ctx.collect(new Tuple3(str[i], temperature, System.currentTimeMillis()));
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        });

        tuple3DataStreamSource.assignTimestampsAndWatermarks(WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(2))
        .withTimestampAssigner(new SerializableTimestampAssigner>() {
            @Override
            public long extractTimestamp(Tuple3 stringIntegerLongTuple3, long l) {
                return stringIntegerLongTuple3.f2;
            }
        })).keyBy(new KeySelector, String>() {
            @Override
            public String getKey(Tuple3 stringIntegerLongTuple3) throws Exception {
                return stringIntegerLongTuple3.f0;
            }
        }).process(new KeyedProcessFunction, String>() {
            Long interval = 10 * 1000l;
            ReducingState reducingState = null;
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                ReducingStateDescriptor reducingStateDescriptor = new ReducingStateDescriptor("reducingState",new Max(),Integer.class);
                reducingState = getRuntimeContext().getReducingState(reducingStateDescriptor);
            }

            @Override
            public void processElement(Tuple3 value, Context ctx, Collector out) throws Exception {
                // 注册定时器10s触发一次,相同定时器重复注册会忽略
                Long statrTimestamp = ctx.timestamp() - (ctx.timestamp() % interval);
                Long timerTimestamp = statrTimestamp + interval;
                ctx.timerService().registerEventTimeTimer(timerTimestamp);

                // 加入新元素
                reducingState.add(value.f1);
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                out.collect("[" + ctx.getCurrentKey() + "] " + "10s内最大温度是" + reducingState.get());
            }

        }).print();

        env.execute("reduceState");


    }

    private static class Max implements ReduceFunction {

        @Override
        public Integer reduce(Integer integer, Integer t1) throws Exception {
            return Math.max(integer,t1);
        }
    }
}