谷歌数据流和发布订阅 - 无法实现一次交付

Question

我正在尝试使用Google Dataflow和Apache Beam SDK 2.6.0的PubSub实现一次性交付。

用例非常简单:

'Generator'数据流作业将1M消息发送到PubSub主题。

GenerateSequence
          .from(0)
          .to(1000000)
          .withRate(100000, Duration.standardSeconds(1L));

“存档”数据流作业从PubSub订阅中读取消息，并保存到Google云存储中。

pipeline
        .apply("Read events",
            PubsubIO.readMessagesWithAttributes()
                // this is to achieve exactly-once delivery
                .withIdAttribute(ATTRIBUTE_ID)
                .fromSubscription('subscription')
                .withTimestampAttribute(TIMESTAMP_ATTRIBUTE))
        .apply("Window events",
            Window.<Dto>into(FixedWindows.of(Duration.millis(options.getWindowDuration())))
                .triggering(Repeatedly.forever(AfterWatermark.pastEndOfWindow()))
                .withAllowedLateness(Duration.standardMinutes(15))
                .discardingFiredPanes())
        .apply("Events count metric", ParDo.of(new CountMessagesMetric()))
        .apply("Write files to archive",
            FileIO.<String, Dto>writeDynamic()
                .by(Dto::getDataSource).withDestinationCoder(StringUtf8Coder.of())
                .via(Contextful.of((msg, ctx) -> msg.getData(), Requirements.empty()), TextIO.sink())
                .to(archiveDir)
                .withTempDirectory(archiveDir)
                .withNumShards(options.getNumShards())
                .withNaming(dataSource ->
                    new SyslogWindowedDataSourceFilenaming(dataSource, archiveDir, filenamePrefix, filenameSuffix)
                ));

我在 Pubsub.IO.Write（“生成器”作业）和 PubsubIO.Read（“存档”作业）中都添加了“带 Id 属性”，并期望它能保证精确一次的语义。

我想测试“负面”情景:

1. “生成器”数据流作业向PubSub主题发送1百万条消息
2. “归档”数据流作业开始工作，但我在处理过程中单击“停止作业”将其停止-

事实上，我得到的是——所有消息都被传递了（至少实现了一次），但除此之外，还有很多重复的消息——大约每1M3050K。

是否有任何解决方案可以实现一次交付？

Answer 1

同时，Pub/Sub支持“恰好一次交付”。

它目前处于正式发布前的状态，因此遗憾的是尚未准备好投入生产使用。

Answer 2

所以，我自己从来没有做过，但是关于你的问题的推理是我处理它的方式…

我的解决方案有点复杂，但我无法在不涉及其他外部服务的情况下确定其他方法来实现这一目标。所以，这里什么都没有。

您可以让您的管道同时从pubsub和GCS读取数据，然后将它们组合起来对数据进行重复删除。这里棘手的部分是一个是有界pCollection（GCS），另一个是无界pCollection（pubsub）。您可以将时间戳添加到有界集合中，然后对数据进行窗口处理。在此阶段，您可能会删除大于约15分钟（您的先例实现中窗口的持续时间）的GCS数据。这两个步骤（即正确添加时间戳和删除可能旧到无法创建重复的数据）是迄今为止最棘手的部分。

解决此问题后，请附加两个 p 集合，然后在两组数据通用的 Id 上使用 GroupByKey。这将产生一个P收集

最后，只有在重新启动管道时，第一个pubsub窗口才需要这项额外的工作。之后，您应该将GCS pCollection重新分配给一个空pCollection，这样按键分组就不会做太多额外的工作。

让我知道你的想法，如果这可行。此外，如果你决定采用这个策略，请公布你的里程数：）。

Answer 3

数据流不能让你在运行中保持状态。如果您使用Java，您可以以不导致其丢失现有状态的方式更新正在运行的管道，从而允许您跨管道版本进行重复数据删除。

如果这对您不起作用，您可能希望以ATTRIBUTE_ID键控的方式存档邮件，例如。使用扳手或GCS作为文件名。