[Flink] aggregation operator

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());

        // Press key groups to perform aggregate query   Extract the latest access data of the current user   max Only take the maximum value of the current specified field  maxBy Round the maximum value of a piece of data 
        SingleOutputStreamOperator<Event> max = dataStream.keyBy(new KeySelector<Event, String>() {

            // Group here by user name 
            @Override
            public String getKey(Event event) throws Exception {
                return event.user;
            }
        }).max("timestamp");

        SingleOutputStreamOperator<Event> maxBy = dataStream.keyBy(new KeySelector<Event, String>() {

            // Group here by user name 
            @Override
            public String getKey(Event event) throws Exception {
                return event.user;
            }
        }).maxBy("timestamp");

        // Print 
        max.print("max");
        maxBy.print("maxBy");

        // Start 
        env.execute();
    }