使用 GroupIntoBatches 转换将元素分组以高效地调用外部服务

通常，编写 Apache Beam 管道可以使用开箱即用的工具和转换，例如 ParDo、Window 和 GroupByKey。但是，当您需要更严格的控制时，您可以在一个原本无状态的 DoFn 中保持状态。

状态是在每个键和每个窗口的基础上保存的，因此，您有状态的 DoFn 的输入需要进行键控（例如，如果您正在跟踪来自电子商务网站的点击次数，则通过客户标识符进行键控）。

用例的示例包括：为每个元素分配一个唯一的 ID，以“更奇特”的方式连接数据流，或者将 API 调用批处理到外部服务。在本节中，我们将重点介绍最后一个用例。

请务必查看 文档，以深入了解状态和计时器。

GroupIntoBatches 转换在后台使用状态和计时器，以允许用户对以下参数进行严格控制

• maxBufferDuration：限制批处理发出等待时间的数量。
• batchSize：限制一个批处理中的元素数量。
• batchSizeBytes： （仅限 Java）限制一个批处理的字节大小（使用输入编码器来确定元素大小）。
• elementByteSize： （仅限 Java）限制一个批处理的字节大小（使用用户定义的函数来确定元素大小）。

同时将实现细节抽象出来，避免用户了解。

withShardedKey() 功能通过将一个键分散到多个线程来提高并行性。

这些转换在 Java 和 Python 中的使用方式如下

input.apply(
          "Batch Contents",
          GroupIntoBatches.<String, GenericJson>ofSize(batchSize)
              .withMaxBufferingDuration(maxBufferingDuration)
              .withShardedKey())

input | GroupIntoBatches.WithShardedKey(batchSize, maxBufferingDuration)

应用这些转换将在每个键的基础上输出一批元素组，然后您可以使用它们以批量方式而不是以每个元素的方式调用外部 API，从而降低管道中的开销。