过滤apache flink中的唯一事件

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class UniqueEventFilter {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建一个数据流
        DataStream<Event> events = env.fromElements(
                new Event("event1", "data1"),
                new Event("event2", "data2"),
                new Event("event1", "data3"),
                new Event("event3", "data4")
        );

        // 定义窗口并应用窗口操作
        DataStream<Event> windowedStream = events
                .keyBy(Event::getEventId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .apply((key, window, input, out) -> {
                    for (Event event : input) {
                        out.collect(event);
                    }
                });

        // 过滤重复事件
        DataStream<Event> uniqueEvents = windowedStream
                .filter(new FilterFunction<Event>() {
                    @Override
                    public boolean filter(Event event) throws Exception {
                        // 根据事件ID判断是否为重复事件
                        // 可以使用状态管理功能来判断事件是否已经存在
                        // 如果事件已经存在，则返回false，过滤掉该事件
                        // 否则返回true，保留该事件
                        // 示例中使用一个HashSet来存储已经处理过的事件ID
                        return processedEventIds.add(event.getEventId());
                    }
                });

        uniqueEvents.print();

        env.execute("Unique Event Filter");
    }

    public static class Event {
        private String eventId;
        private String eventData;

        public Event(String eventId, String eventData) {
            this.eventId = eventId;
            this.eventData = eventData;
        }

        public String getEventId() {
            return eventId;
        }

        public String getEventData() {
            return eventData;
        }
    }
}