3

目的：

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

问题：

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

如果我减小元素大小（元素数量不会改变），它会起作用！因为实际上 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. 应用带触发的窗口（在行较小且没有内存问题的情况下会触发大约 50 次）
  7. 每个键计数（按键分组，然后组合它们）
  8. 最后，我们应该有大约 50 * 10 个元素来表示按窗口和键的计数（当行大小足够小时成功测试）

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

在此处输入图像描述

按键分组步骤摘要：

在此处输入图像描述

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

窗口代码在这里：

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 发出每个窗口的早期结果来解决内存问题。

4

1 回答 1

5

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

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

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

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

于 2019-04-16T20:00:29.610 回答