Flink GlobalWindow触发仅处理触发事件，如何实现全事件处理？

问题场景 #

将数据流按事件属性执行keyBy操作后传入GlobalWindow，设置当特定事件（如eod事件）到来时触发窗口处理，但触发时仅处理该触发事件，无法获取窗口内所有事件。

测试输入事件：

test_one, one
event_two, two
event_three, three
event_four, four
eod, eod

实际输出仅为：Key: eod, Value: eod，期望输出所有发送的事件。

用户提供的原始代码 #

public class MyEvent {
    public String key;
    public String value;

    public MyEvent() {
        // Default constructor
    }

    public MyEvent(String key, String value) {
        this.key = key;
        this.value = value;
    }

    public String getKey() {
        return key;
    }

    public void setKey(String key) {
        this.key = key;
    }

    public String getValue() {
        return value;
    }

    public void setValue(String value) {
        this.value = value;
    }

    @Override
    public String toString() {
        return "Key: " + key + ", Value: " + value;
    }
}

public class KeyedGlobalWindowTriggerExample {

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

        DataStream<MyEvent> input = env.socketTextStream("localhost", 9091)
                .map(new MapFunction<String, MyEvent>() {
                    @Override
                    public MyEvent map(String value) {
                        // Assuming the input stream is in the format "key,value"
                        String[] parts = value.split(",");
                        return new MyEvent(parts[0], parts[1]);
                    }
                });

        // Key By Event Property
        KeyedStream<MyEvent, String> keyedStream = input
                .keyBy(event -> event.key);

        //Create a Custom Trigger
        keyedStream.window(GlobalWindows.create())
                .trigger(new Trigger<MyEvent, GlobalWindow>() {
                    @Override
                    public TriggerResult onElement(MyEvent event, long timestamp, GlobalWindow window, TriggerContext ctx) {
                        if ("eod".equals(event.getKey())) {
                            return TriggerResult.FIRE;
                        }
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) {
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) {
                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public void clear(GlobalWindow window, TriggerContext ctx) {
                        // Handle clearing of window state if necessary
                    }
                })
                .process(new MyProcessWindowFunction())
                .print();

        env.execute("Keyed Global Window Trigger Example");
    }
}

public class MyProcessWindowFunction extends ProcessWindowFunction<MyEvent, String, String, GlobalWindow> {
    @Override
    public void process(String key, Context context, Iterable<MyEvent> elements, Collector<String> out) {
        for (MyEvent element : elements) {
            out.collect(element.toString());
        }
    }
}

核心原因 #

keyBy(event -> event.key)会将每个不同的key分配到独立的GlobalWindow：

• 前四个事件的key分别为test_one、event_two、event_three、event_four，它们各自对应一个独立窗口，但这些窗口没有触发信号（没有对应key的eod事件），因此永远不会被处理。
• eod事件的key是eod，它的窗口中只有自身，触发时自然只能输出自己。

解决方案 #

方案1：所有事件进入同一窗口，等待全局eod触发 #

如果需要将所有事件汇总到一个窗口，由全局eod事件触发处理，修改keyBy逻辑，使用固定key将所有事件分到同一窗口：

// 修改keyBy逻辑，所有事件共用同一个key
KeyedStream<MyEvent, String> keyedStream = input
        .keyBy(event -> "global_key");

同时调整触发器，使用FIRE_AND_PURGE触发后清空窗口，避免重复处理：

.trigger(new Trigger<MyEvent, GlobalWindow>() {
    @Override
    public TriggerResult onElement(MyEvent event, long timestamp, GlobalWindow window, TriggerContext ctx) {
        if ("eod".equals(event.getKey())) {
            // 触发窗口并清空，防止后续事件重复处理
            return TriggerResult.FIRE_AND_PURGE;
        }
        // 非eod事件存入窗口等待触发
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, GlobalWindow window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, GlobalWindow window, TriggerContext ctx) {
        return TriggerResult.CONTINUE;
    }

    @Override
    public void clear(GlobalWindow window, TriggerContext ctx) {
    }
})

此时发送测试事件，会输出所有积累的事件，然后清空窗口。

方案2：按业务key分组，每个组等待自身eod触发 #

如果需要按业务key分组，每个组由对应key的eod事件触发（比如test_one组的eod事件为test_one,eod），修改触发器逻辑，通过value判断是否为触发信号：

.trigger(new Trigger<MyEvent, GlobalWindow>() {
    @Override
    public TriggerResult onElement(MyEvent event, long timestamp, GlobalWindow window, TriggerContext ctx) {
        // 当事件value为eod时，触发当前key的窗口
        if ("eod".equals(event.getValue())) {
            return TriggerResult.FIRE_AND_PURGE;
        }
        return TriggerResult.CONTINUE;
    }

    // 其他方法保持不变
})

测试输入示例：

test_one, one
test_one, two
test_one, eod
test_two, three
test_two, eod

此时test_one组会在收到test_one,eod时触发，输出该组的所有事件；test_two组同理。

内容的提问来源于stack exchange，提问作者car_dev