大数据-125 - Flink State 02篇 状态原理和原理剖析：广播状态

DataStream<Action> actions = ???`
`DataStream<Pattern> patterns = ???
Action``Pattern

package icu.wzk;


import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
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.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;


public class BroadCastDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        // 两套数据流 1:用户行为 2:模式
        UserAction user1 = new UserAction(1001L, "login");
        UserAction user2 = new UserAction(1003L, "pay");
        UserAction user3 = new UserAction(1002L, "car");
        UserAction user4 = new UserAction(1001L, "logout");
        UserAction user5 = new UserAction(1003L, "car");
        UserAction user6 = new UserAction(1002L, "logout");
        DataStreamSource<UserAction> actions = env.fromElements(user1, user2, user3, user4, user5, user6);
        MyPattern myPattern1 = new MyPattern("login", "logout");
        MyPattern myPattern2 = new MyPattern("car", "logout");
        DataStreamSource<MyPattern> patterns = env.fromElements(myPattern1, myPattern2);
        KeyedStream<UserAction, Long> keyed = actions
                .keyBy(new KeySelector<UserAction, Long>() {
                    @Override
                    public Long getKey(UserAction value) throws Exception {
                        return value.getUserId();
                    }
                });
        // 将模式流广播到下游的所有算子
        MapStateDescriptor<Void, MyPattern> broadcastStateDescriptor = new MapStateDescriptor<>(
                "patterns", Types.VOID, Types.POJO(MyPattern.class));
        BroadcastStream<MyPattern> broadcastPatterns = patterns.broadcast(broadcastStateDescriptor);

        SingleOutputStreamOperator<Tuple2<Long, MyPattern>> process = keyed
                .connect(broadcastPatterns)
                .process(new PatternEvaluator());

        // 匹配结果输出到控制台
        process.print();
        env.execute("BroadCastDemo");
    }

    // 用户行为类
    public static class UserAction {
        private Long userId;
        private String userAction;

        public UserAction(Long userId, String userAction) {
            this.userId = userId;
            this.userAction = userAction;
        }

        public Long getUserId() {
            return userId;
        }

        public String getUserAction() {
            return userAction;
        }
    }

    // 模式类
    public static class MyPattern {
        private String firstAction;
        private String secondAction;

        // 无参构造函数
        public MyPattern() {
        }

        public MyPattern(String firstAction, String secondAction) {
            this.firstAction = firstAction;
            this.secondAction = secondAction;
        }

        public String getFirstAction() {
            return firstAction;
        }

        public void setFirstAction(String firstAction) {
            this.firstAction = firstAction;
        }

        public String getSecondAction() {
            return secondAction;
        }

        public void setSecondAction(String secondAction) {
            this.secondAction = secondAction;
        }
    }

    public static class PatternEvaluator extends KeyedBroadcastProcessFunction<Long, UserAction, MyPattern, Tuple2<Long, MyPattern>> {

        private transient ValueState<String> prevActionState;

        @Override
        public void open(Configuration parameters) throws Exception {
            prevActionState = getRuntimeContext()
                    .getState(new ValueStateDescriptor<>("lastAction", Types.STRING));
        }

        /**
         * 每个一个Action数据，触发一次执行
         * @author wzk
         * @date 11:21 2024/7/29
        **/
        @Override
        public void processElement(UserAction value, KeyedBroadcastProcessFunction<Long, UserAction, MyPattern, Tuple2<Long, MyPattern>>.ReadOnlyContext ctx, Collector<Tuple2<Long, MyPattern>> out) throws Exception {
            // 把用户行为流和模式流中的模式进行匹配
            ReadOnlyBroadcastState<Void, MyPattern> patterns = ctx
                    .getBroadcastState(new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(MyPattern.class)));
            MyPattern myPattern =  patterns.get(null);
            String prevAction = prevActionState.value();
            if (myPattern != null && prevAction != null) {
                if (myPattern.getFirstAction().equals(prevAction) && myPattern.getSecondAction().equals(value.getUserAction())) {
                    // 匹配成功
                    System.out.println("匹配成功: " + ctx.getCurrentKey());
                    out.collect(new Tuple2<>(ctx.getCurrentKey(), myPattern));
                } else {
                    // 匹配失败
                    System.out.println("匹配失败: " + ctx.getCurrentKey());
                }
            }
            prevActionState.update(value.getUserAction());
        }

        /**
         * 每次来一个模式 Pattern 的时候触发
         * @author wzk
         * @date 11:29 2024/7/29
        **/
        @Override
        public void processBroadcastElement(MyPattern value, KeyedBroadcastProcessFunction<Long, UserAction, MyPattern, Tuple2<Long, MyPattern>>.Context ctx, Collector<Tuple2<Long, MyPattern>> out) throws Exception {
            BroadcastState<Void, MyPattern> broadcastState = ctx
                    .getBroadcastState(new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(MyPattern.class)));
            broadcastState.put(null, value);
        }
    }

}