Flink MapState实例

MapState的方法和Java的Map的方法极为相似，所以上手相对容易。
常用的有如下：

• get()方法获取值
• put()，putAll()方法更新值
• remove()删除某个key
• contains()判断是否存在某个key
• isEmpty() 判断是否为空

需求：统计每个用户的行为次数

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
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.datastream.SingleOutputStreamOperator;
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 MapStateTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().setAutoWatermarkInterval(100);
        DataStreamSource> tuple2DataStreamSource = env.addSource(new SourceFunction>() {
            boolean flag = true;

            @Override
            public void run(SourceContext> sourceContext) throws Exception {
                String[] s = {"张三", "王五", "李四", "秋英"};
                String[] s1 = {"登录", "退出", "加购", "够买"};
                while (flag) {
                    Thread.sleep(1000);
                    int i = new Random().nextInt(4);
                    sourceContext.collect(new Tuple3(s[i], s1[i], System.currentTimeMillis()));
                }
            }

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

        SingleOutputStreamOperator> tuple3SingleOutputStreamOperator = tuple2DataStreamSource.assignTimestampsAndWatermarks(WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner>() {
                    @Override
                    public long extractTimestamp(Tuple3 stringLongTuple2, long l) {
                        return stringLongTuple2.f2;
                    }
                }));

        tuple3SingleOutputStreamOperator.keyBy(new KeySelector, String>() {
            @Override
            public String getKey(Tuple3 stringStringLongTuple3) throws Exception {
                return stringStringLongTuple3.f0;
            }
        }).process(new KeyedProcessFunction, String>() {

            MapState mapState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                MapStateDescriptor mapStateDescriptor = new MapStateDescriptor("mapstate",String.class,Integer.class);
                mapState = getRuntimeContext().getMapState(mapStateDescriptor);
            }

            @Override
            public void processElement(Tuple3 value, Context ctx, Collector out) throws Exception {
                // 初始化
                if(!mapState.contains(value.f1)){
                    mapState.put(value.f1,1);
                }

                // "登录", "退出", "加购", "够买"
                mapState.put(value.f1,mapState.get(value.f1) + 1);

                out.collect(value.f0 + "[登录次数：" + mapState.get("登录") + ",退出次数：" + mapState.get("退出") + ",加购次数：" + mapState.get("加购") + ",够买次数：" + mapState.get("够买") + "]");

            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector out) throws Exception {
                super.onTimer(timestamp, ctx, out);
            }

        }).print();

        env.execute("mapState");

    }
}