谷歌云数据流工作线程

Question

假设我们有一个有 4 个 CPU 内核的工人。Dataflow 工作机器中的并行性如何配置？我们是否并行超过内核数量？

在哪里可以获得此类信息？

Answer 1

我之前还认为数据流工作人员每个核心使用 1 个线程。但是，我最近发现这仅适用于批处理模式。在流模式，除非另有规定，它使用300个线程所看到这里。这与批处理工作代码相反。要限制工作线程的数量，请使用--numberOfWorkerHarnessThreads=N.

我如何发现这一点的故事：我有一个带有 ParDo 的流媒体作业，可以读取 XML 文件并解析它们。工作人员内存不足，因为他们试图一次处理太多文件。我使用静态 AtomicInteger 来计算 ParDo 的并发执行次数。此外，我还记录了用于执行 ParDo 的 threadId。我可以看到一个 worker 上有多达 300 个并发执行，并且记录的不同 threadId 的数量也表明该 worker 正在使用大量线程。

我通过限制与--numberOfWorkerHarnessThreads=10. 我还尝试将该数字设置为 1 个线程，但这似乎导致在任何给定时间只执行一个管道步骤。不足为奇，但我想要更高级别的并行性，因此 10 对我的用例来说似乎是一个不错的数字。

编辑：添加了一些额外的信息。感谢 @safurudin-mahic 找到相关的 Beam 源代码。