带窗口的 GroupByKey 后，Beam 管道不产生任何输出，并且出现内存错误

Question

目的：

我想加载流数据，然后添加一个键，然后按键计数。

问题：

当我尝试使用流方法（无界数据）加载和按键分组大数据时，Apache Beam Dataflow 管道出现内存错误。因为似乎数据是按分组累积的，并且不会在触发每个窗口时更早地触发数据。

如果我减小元素大小（元素数量不会改变），它会起作用！因为实际上 group-by step 等待所有数据被分组，然后触发所有新的窗口数据。

我对两者进行了测试：

梁版本 2.11.0 和 scio 版本 0.7.4

梁版本 2.6.0 和 scio 版本 0.6.1

重新生成错误的方法：

1. 阅读包含文件名的 Pubsub 消息
2. 从 GCS 读取并加载相关文件作为逐行迭代器
3. 逐行展平（因此它生成大约 10,000 个）元素
4. 向元素添加时间戳（当前即时时间）
5. 创建我的数据的键值（使用一些从 1 到 10 的随机整数键）
6. Apply window with triggering（在行很小且没有内存问题的情况下会触发大约50次）
7. 每个键计数（按键分组然后组合它们）
8. 最后，我们应该有大约 50 * 10 个元素代表按窗口和键的计数（当行大小足够小时测试成功）

管道的可视化（步骤 4 到 7）：

group-by-key 步骤的总结：

如您所见，数据是按组累积的，不会发出。

窗口代码在这里：

val windowedData = data.applyKvTransform(
  Window.into[myt](
    Sessions.withGapDuration(Duration.millis(1)))
    .triggering(
      Repeatedly.forever(AfterFirst.of(
        AfterPane.elementCountAtLeast(10),
        AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.millis(1)))

      ).orFinally(AfterWatermark.pastEndOfWindow())

    ).withAllowedLateness(Duration.standardSeconds(100))
    .discardingFiredPanes()

)

错误：

org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$KeyCommitTooLargeException: Commit request for stage S2 and key 2 is larger than 2GB and cannot be processed. This may be caused by grouping a very large amount of data in a single window without using Combine, or by producing a large amount of data from a single input element.
    org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$KeyCommitTooLargeException.causedBy(StreamingDataflowWorker.java:230)
    org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1287)
    org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.access$1000(StreamingDataflowWorker.java:146)
    org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$6.run(StreamingDataflowWorker.java:1008)
    java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    java.lang.Thread.run(Thread.java:745)

是否有任何解决方案可以通过强制 group-by 发出每个窗口的早期结果来解决内存问题。

Answer 1

KeyCommitTooLargeException 不是内存问题，而是 protobuf 序列化问题。Protobuf 的对象限制为 2GB（google protobuf 最大大小）。Dataflow 发现管道中单个键的值大于 2GB，因此无法对数据进行 shuffle。错误消息表明“这可能是由于在没有使用组合的情况下在单个窗口中对大量数据进行分组，或者从单个输入元素生成大量数据造成的。” 根据您的管道设置（即分配的随机密钥），后者更有可能。

管道可能从 GCS 读取了一个大文件 (>2GB) 并将其分配给一个随机密钥。GroupByKey 需要密钥 shuffle 操作，而 Dataflow 由于 protobuf 限制而无法执行，因此卡在该密钥上并保留水印。

如果单个键的值较大，您可能希望减小值的大小，例如，压缩字符串，或将字符串拆分为多个键，或者首先生成较小的 GCS 文件。

如果较大的值来自多个键的分组，您可能希望增加键空间，以便按键操作的每个组最终将较少的键组合在一起。