flink使MapState实现KeyedState

测试数据

辽宁省,沈阳市,1000
辽宁省,大连市,2000
辽宁省,沈阳市,1500
湖南省,长沙市,1200
湖南省,长沙市,1000
湖南省,常德市,4000
湖南省,常德市,3000

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.time.Time;
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.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

/** * @Author: Zhang * @Description: * @Date: Created in 19:23 2021/12/25 * @Modified By: * */
public class MapStateDemo {
    

    public static void main(String[] args) throws Exception {
    


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //开启checkpoint, 每10s进行一次checkpoint, 开启后默认使用无限重启策略
        env.enableCheckpointing(10000);

        //开启失败率重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(
                3, // max failures per unit
                Time.of(30, TimeUnit.SECONDS), //time interval for measuring failure rate
                Time.of(3, TimeUnit.SECONDS) // delay

        ));

        DataStreamSource<String> lines = env.socketTextStream("doitedu03", 8888);


        SingleOutputStreamOperator<Tuple3<String,String,Integer>> tpStream = lines.flatMap(new FlatMapFunction<String, Tuple3<String,String,Integer>>() {
    
            @Override
            public void flatMap(String value, Collector<Tuple3<String,String,Integer>> collector) throws Exception {
    
                String[] sp = value.split(",");
                if (sp[0].equals("error")){
    
                    throw  new RuntimeException("出现错误了！！！");
                }
                collector.collect(Tuple3.of(sp[0],sp[1],Integer.parseInt(sp[2])));
            }
        });

        KeyedStream<Tuple3<String, String, Integer>, String> keyed = tpStream.keyBy(t -> t.f0);


        SingleOutputStreamOperator<Tuple3<String, String, Integer>> process = keyed.process(new KeyedProcessFunction<String, Tuple3<String, String, Integer>, Tuple3<String, String, Integer>>() {
    


            private transient MapState<String, Integer> mapState;

            @Override
            public void open(Configuration parameters) throws Exception {
    
                //先定义一个状态描述器
                MapStateDescriptor<String, Integer> mapStateDescriptor = new MapStateDescriptor<>("map-state", String.class, Integer.class);
                //初始化或者恢复历史状态
                mapState = getRuntimeContext().getMapState(mapStateDescriptor);
            }

            @Override
            public void processElement(Tuple3<String, String, Integer> input, Context context, Collector<Tuple3<String, String, Integer>> collector) throws Exception {
    

                String city = input.f1;
                Integer money = input.f2;
                Integer historyMoney = mapState.get(city);
                if (historyMoney == null) {
    
                    historyMoney = 0;
                }
                Integer totalMoney = money + historyMoney;
                //更新到state中
                mapState.put(city, totalMoney);
                //输出
                input.f2 = totalMoney;
                collector.collect(input);
            }
        });

        process.print();

        env.execute();


    }
}