用Apache Beam窗口-固定窗口似乎不关闭？

Question

我们试图在Apache Beam管道上使用固定窗口（使用DirectRunner）。我们的流程如下：

1. 从pub/sub中提取数据
2. 反序列化JSON到Java对象
3. 窗口事件w/固定窗口5秒
4. 使用自定义的CompineFn，将事件的每个窗口合并成一个List

管道代码：

    pipeline
                // Read from pubsub topic to create unbounded PCollection
                .apply(PubsubIO
                    .<String>read()
                    .topic(options.getTopic())
                    .withCoder(StringUtf8Coder.of())
                )

                // Deserialize JSON into Event object
                .apply("ParseEvent", ParDo
                    .of(new ParseEventFn())
                )

                // Window events with a fixed window size of 5 seconds
                .apply("Window", Window
                    .<Event>into(FixedWindows
                        .of(Duration.standardSeconds(5))
                    )
                )

                // Group events by window
                .apply("CombineEvents", Combine
                    .globally(new CombineEventsFn())
                    .withoutDefaults()
                )

                // Log grouped events
                .apply("LogEvent", ParDo
                    .of(new LogEventFn())
                );

我们看到的结果是，最后一步永远不会运行，因为我们没有任何日志记录。

此外，我们还添加了系统。出来println（“***”）在我们自定义的CombineFn类的每个方法中，以便跟踪这些方法何时运行，而它们似乎也不运行。

这里的窗口设置有误吗？我们遵循了https://beam.apache.org/documentation/programming-guide/#windowing的一个例子，它看起来相当简单，但显然缺少一些基本的东西。

任何见解都值得赞赏-提前感谢！




Answer 1

看起来主要问题确实是一个丢失的触发器——窗口正在打开，没有什么告诉它什么时候发出结果。我们想简单地根据流转时长（而不是事件时间）进行窗口，以下内容也是如此：

.apply("Window", Window
    .<Event>into(new GlobalWindows())
    .triggering(Repeatedly
        .forever(AfterProcessingTime
            .pastFirstElementInPane()
            .plusDelayOf(Duration.standardSeconds(5))
        )
    )
    .withAllowedLateness(Duration.ZERO).discardingFiredPanes()
)

本质上，这创建了一个全局窗口，在处理第一个元素5秒后触发该窗口以发出事件。每次窗口关闭时，另一个窗口在接收到元素后打开。Beam抱怨我们没有允许延迟的，片段-据我所知，这只是告诉它忽略任何延迟数据。

在这里，我的理解可能有点离题，但上面的代码片段解决了我们的问题！