Apache Beam 在相同数量的元素后触发窗格

Question

我们使用 Apache Beam 向 API 发送一些数据。对于单个 API 调用，API 仅接受一定数量的元素。

为了将元素分组为微批次，我们使用了混合触发器，该触发器在以下条件之一发生时触发：1) 30 秒后 2) 当至少收集 10 个元素时。

现在看来，在某些情况下，触发器会为单个窗格传递超过 10 个元素。有什么建议可以强制记录的确切数量吗？我想，正如方法名称所示，可以传递 '.elementsCountAtLeast(int x)

more then

x` 元素。

// definition
public class WindowHelper {

    public static Window<KV<String,String>> generateWIndow() {
        Window <KV<String,String>> window = Window.<KV<String,String>>into(
            new GlobalWindows()).triggering(
                Repeatedly.forever(AfterFirst.of(
                    AfterPane.elementCountAtLeast(10),
                    AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardSeconds(30))))
    ).discardingFiredPanes();

    }
}

// invocation
data.apply(WindowHelper.generateWindow())

...

Answer 1

我会回想一下你为什么想要这个，在两个选项之间做出决定：

根据设计，您
• 无法使用触发器来获取准确数量的元素：为了提高效率，Beam runner 通过一起处理元素束来摊销成本。这不仅适用于批处理，也适用于流式传输。因此触发器会说“好吧，是时候输出了”，但跑步者总是可以等待，直到“好”时间。这通常是在处理包结束时，所有数据都将被刷新。
您
• 可以使用GroupIntoBatches来批量获取小于或等于 10个元素。最后一批可能很小，冲完不到10个。

现在，在流处理中，您可能经常返回并重新处理旧数据以修复错误或运行实验。您应该考虑这一点以帮助您做出决定。

重要：

对于
• GroupIntoBatches

  ，批次的大小是输入/输出规范的一部分。在流式处理和批处理场景中，您都将获得 10 个或更少元素的批次。

对于触发器来说，它
• 只是用于平衡成本/延迟/结果完整性的控制。在这种情况下，跑步者可能会产生巨大束并仅在最后触发。