有状态索引导致 ParDo 在 Dataflow Runner 上单线程运行

Question

我们正在使用 Beam 的 Java SDK 2.0.0 在 ParDo 中生成顺序索引。就像 Beam 的 introduction to stateful processing 中的简单状态索引示例一样，我们使用 ValueState<Integer> 单元格，我们对它的唯一操作是检索值并在需要下一个索引时递增：

Integer statefulIndex = firstNonNull(index.read(), 0);
index.write(statefulIndex + 1);

使用 Google 的 Dataflow runner 运行时，我们在 Dataflow 监控界面上注意到 ParDo 的挂墙时间与经过的时间同步累积。我们能够通过 ssh 进入工作节点并使用 top 和 1 查看每个内核的 CPU 使用率来确认 ParDo 执行单线程。注释掉有状态处理单元并保持代码不变，同一个 ParDo 使用我们 n1-standard-32 工作节点的所有核心。

即使 Dataflow 运行器能够基于每个键和窗口对并行化状态索引（我们目前有一个窗口和一个键），但缺乏并行性会导致性能显着下降，以至于我们无法使用它.这是 Dataflow 运行器的预期行为吗？

天真地，我预计 Beam 的状态索引在幕后会像 Java 的 AtomicInteger 那样运行。是否存在阻止使用 ValueState<Integer> 单元进行并行处理的限制，或者此功能尚未内置到运行器中？

Answer 1

这不仅是 Dataflow 运行器的预期行为，而且在任何情况下都是逻辑必要的。无论您是在 Beam 中使用 state 还是在单进程 Java 程序中使用 AtomicInteger 都没有关系：如果操作“A”写入一个值，而操作“B”读取该值，那么“B”必须在之后执行“一种”。这种关系的常用术语是“发生在之前”。

这种形式的有状态计算与并行计算相反。根据定义，观察到写入的读取具有因果关系。根据定义，两个并行的操作没有因果关系。

现在，您可能期望并行线程同时访问状态单元，就像在具有一些共享状态和并发控制的多线程编程的标准模式中一样。对于此示例，如果这些线程实际上是并行的，您将获得重复的索引。退后一步，Beam 的目标是透明地分布在大型机器集群中的大规模“令人尴尬的并行”计算。细粒度的并发控制除了极难正确之外，还不容易转化为大规模的分布式计算。