谷歌云数据流因工作人员失联导致组合功能失效

Question

我的数据流在我的组合函数中始终失败，除了以下条目之外，日志中没有报告错误：

 A work item was attempted 4 times without success. Each time the worker eventually lost contact with the service.

我使用的是Apache Beam Python SDK 2.4.0。我已经尝试使用CombinePerKey和CombineGlobally执行此步骤。在这两种情况下，管道在组合函数中都失败了。当使用较小的数据量运行时，管道完成。

使用n1-highmem-4workers也会出现同样的失败。当我检查Stackdriver时，我没有看到错误，但是有三种警告:找不到会话文件、拒绝拆分和处理停顿。我的输入集合大小显示，它有17000个元素，分布在60 MB，但Stackdriver有一个声明，说我在一个工人身上使用了25 GB，这正在接近最大值。对于这个输入，在我的组合中创建的每个累加器应该占用大约150 MB内存。我的管道是否创建了太多累加器，耗尽了它的内存？如果是，我如何告诉它更频繁地合并累加器或限制创建的数量？

我确实有一个错误日志条目来验证我的工人是由于OOM而死亡的。它只是没有标记为工作者错误，这是数据流监视器的默认筛选。

管道定义类似于：

table1 = (p | "Read Table1" >> beam.io.Read(beam.io.BigQuerySource(query=query))
     | "Key rows" >> beam.Map(lambda row: (row['key'], row)))
table2 = (p | "Read Table2" >> beam.io.Read(beam.io.BigQuerySource(query=query))
     | "Key rows" >> beam.Map(lambda row: (row['key'], row)))

merged = ({"table1": table1, "table2": table2}
     | "Join" >> beam.CoGroupByKey()
     | "Reshape" >> beam.ParDo(ReshapeData())
     | "Key rows" >> beam.Map(lambda row: (row['key'], row)))
     | "Build matrix" >> beam.CombinePerKey(MatrixCombiner())  # Dies here
     | "Write matrix" >> beam.io.avroio.WriteToAvro())

Answer 1

用更少的工人运行导致更少的蓄能器和管道的成功完成。