[Flink] protocol operator reduce

One 、 Entity class

@Data
public class Event {
    public String user;
    public String url;
    public Long timestamp;
    
    public Event(String user, String url, Long timestamp) {
        this.user = user;
        this.url = url;
        this.timestamp = timestamp;
    }
}

Two 、 Custom data sources simulate streaming data

public class ClickSource implements SourceFunction<Event> {
 
    // Declare a flag class 
    private Boolean running = true;
 
    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        // Randomly generated data 
        Random random = new Random();
 
        // Define the data set selected by the field 
        String[] users = {"Mary","Alice","Bob","Cary"};
        String[] urls = {"./home","./cart","./fav","./prod?id=100","./prod:id=10"};
 
        // Loop data 
        while (running){
            String user = users[random.nextInt(users.length)];
            String url = urls[random.nextInt(user.length())];
            Long time = System.currentTimeMillis();
            ctx.collect(new Event(user,url,time));
            Thread.sleep(2000);
        }
    }
 
    @Override
    public void cancel() {
        running = false;
    }
}

3、 ... and 、 Task code

    public static void main(String[] args) throws Exception {
        // Create an execution environment 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Set parallelism 
        env.setParallelism(1);

        // Read data source 
        DataStreamSource<Event> dataStream = env.addSource(new ClickSource());

        // Count the access frequency of each user 
        SingleOutputStreamOperator<Tuple2<String, Integer>> map = dataStream.map(new MapFunction<Event, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Event event) throws Exception {
                return Tuple2.of(event.user, 1);
            }
        });

        // Group first   Group by user name 
        KeyedStream<Tuple2<String, Integer>, String> keyBy = map.keyBy(data -> data.f0);

        // Make a protocol 
        SingleOutputStreamOperator<Tuple2<String, Integer>> reduce = keyBy.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) throws Exception {
                return new Tuple2<>(t1.f0, t1.f1 + t2.f1);
            }
        });

        // Select the most active user currently   To specify a key Put all the data in the same group 
        SingleOutputStreamOperator<Tuple2<String, Integer>> operator = reduce.keyBy(data -> "key")
                .reduce(new ReduceFunction<Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) throws Exception {
                        return t1.f1 > t2.f1 ? t1 : t2;
                    }
                });

        // Print 
        operator.print();

        // Start 
        env.execute();
    }