java - 云数据流 Watermark 卡住并增加系统滞后

Question

我正在从数据流管道中的 PubSub 主题读取记录。 PubSub 记录分为固定窗口，然后在每个窗口上分组。每个窗口都按序列号排序，因为我们需要使用 beam.SortValues 按顺序处理这些记录。然后我将记录写入 Cloud BigTable

管道的问题是数据新鲜度和系统滞后。数据新鲜度似乎停留在某个点，水印停止前进。

我正在使用以下窗口策略在 GroupByKey 步骤后发出记录:

PCollection<KV<BigInteger, JSONObject>> window = pubsubRecords.apply("Raw to String", ParDo.of(new LogsFn()))
                .apply("Window", Window
                        .<KV<BigInteger, JSONObject>>into(FixedWindows.of(Duration.standardSeconds(10)))
                        .triggering(Repeatedly.forever(AfterFirst.of(
      AfterPane.elementCountAtLeast(500),
      AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardMinutes(1)))))
                        .withAllowedLateness(Duration.ZERO).discardingFiredPanes()
                    );

我认为问题可能出在窗口策略上。基本上我想做以下事情:将记录从 PubSub 读取到 1 分钟的 FixedWindows 中，对窗口进行排序并写入 BigTable。如果我使用默认触发器，则 GroupByKey 步骤不会发出任何结果。有人可以帮我解决这个问题吗？

Answer 1

阅读您的代码，现在您的早期触发器和窗口大小似乎落后了。您的开窗策略实际上是:

1. 10 秒事件时间固定窗口
2. 1 分钟处理时间或 Pane 中 500 个元素的复合早期触发器。
3. 丢弃迟到的事件。

如果您只需要 1 分钟的事件时间窗口，则需要以下内容:

PCollection<KV<BigInteger, JSONObject>> window = pubsubRecords.apply("Raw to String", ParDo.of(new LogsFn()))
            .apply("Window", Window
            .<KV<BigInteger, JSONObject>>into(FixedWindows.of(Duration.standardMinutes(1)))
                .withAllowedLateness(Duration.ZERO)
                .discardingFiredPanes()
                .withOnTimeBehavior(OnTimeBehavior.FIRE_ALWAYS));

Fire 始终是默认的 OnTimeBehavior，但我们可以将其显式化以提高可读性。如果您需要复合触发器，您可以将其重新添加 - 我怀疑您想要触发一个 10 秒或 500 个元素。