Apache Beam：具有无限源的批处理管道

我目前正在使用 Apache Beam 和 Google Dataflow 来处理实时数据。数据来自Google PubSub，它是无限制的，所以目前我正在使用流媒体管道。然而，事实证明，拥有一个 24/7 运行的流管道是相当昂贵的。为了降低成本，我正在考虑切换到以固定时间间隔（例如每 30 分钟）运行的批处理管道，因为对于用户来说实时处理并不重要。

我想知道是否可以使用 PubSub 订阅作为有限源？我的想法是，每次运行作业时，都会在触发之前积累1分钟的数据。到目前为止，这似乎不可能，但我遇到过一个名为BoundedReadFromUnboundedSource https://beam.apache.org/documentation/sdks/javadoc/2.2.0/org/apache/beam/sdk/io/BoundedReadFromUnboundedSource.html（我不知道如何使用），所以也许有办法吗？

下面是源的大致样子：

PCollection<MyData> data = pipeline
            .apply("ReadData", PubsubIO
                    .readMessagesWithAttributes()
                    .fromSubscription(options.getInput()))
            .apply("ParseData", ParDo.of(new ParseMyDataFn()))
            .apply("Window", Window
                    .<MyData>into(new GlobalWindows())
                    .triggering(Repeatedly
                            .forever(AfterProcessingTime
                                    .pastFirstElementInPane()
                                    .plusDelayOf(Duration.standardSeconds(5))
                            )
                    )
                    .withAllowedLateness(Duration.ZERO).discardingFiredPanes()
            );

我尝试执行以下操作，但作业仍然以流模式运行：

PCollection<MyData> data = pipeline
            .apply("ReadData", PubsubIO
                    .readMessagesWithAttributes()
                    .fromSubscription(options.getInput()))
            .apply("ParseData", ParDo.of(new ParseMyDataFn()))

            // Is there a way to make the window trigger once and turning it into a bounded source?
            .apply("Window", Window
                    .<MyData>into(new GlobalWindows())
                    .triggering(AfterProcessingTime
                        .pastFirstElementInPane()
                        .plusDelayOf(Duration.standardMinutes(1))
                    )
                    .withAllowedLateness(Duration.ZERO).discardingFiredPanes()
            );

这在中没有明确支持PubsubIO目前，您可以尝试定期启动流作业并在几分钟后以编程方式调用 Drain。