[Flink] transform operator flatmap

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 class TransFormFlatMapTest {
    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());

        // Perform a conversion operation   Flat mapping 
        SingleOutputStreamOperator<String> flatMap = dataStream.flatMap(new FlatMapFunction<Event, String>() {
            @Override
            public void flatMap(Event event, Collector<String> collector) throws Exception {
                if (event.user.equals("Bob")) {
                    collector.collect(event.url);
                }
            }
        });

        // Print 
        flatMap.print();

        // Start 
        env.execute();
    }
}