Google云数据流工作线程

Answer 1

我以前也认为dataflow工作人员使用每个核心1个线程。但是，我最近发现这只适用于批处理模式。在流模式下，除非另有规定，否则它使用300个线程，如下所示。这与批处理工作者代码形成对比。要限制工作线程的数量，请使用--numberofworkerharnessthreads=n。

我的故事是如何发现这一点的：我有一个ParDo的流式工作，它将读取XML文件并解析它们。工人们的内存耗尽了，因为他们试图一次啃太多的文件。我使用一个静态AtomicInteger来计算PARDO的并发执行次数。此外，我还记录了用于执行PARDO的threadIds。我可以在一个worker上看到多达300个并发执行，记录的不同threadid的数量也表明该worker使用了大量线程。

我通过限制--numberofworkerharnessthreads=10使用的线程数来解决内存问题。我还尝试将这个数字设置为1个线程，但这似乎导致在任何给定时间只执行一个流水线步骤。这并不奇怪，但是我想要更高级别的并行性，所以10对于我的用例来说似乎是一个很好的数字。