数据流管道在以低 CPU 利用率缩减后暂停
Dataflow pipeline pauses after scaling down with low CPU utilization
在我们的流媒体管道中，我们从 pubsub 读取数据，进行一些验证，然后在 10 秒间隔会话 window 中按键对其进行分组。之后数据被进一步处理并再次写入bigtable和pubsub。
我们使用的是 apache beam 2.28 和数据流流引擎。白天我们处理的数据比晚上多，管道会自动增加工作人员 (n2d-standard-4) 的数量。大多数情况下，它会从 2 名工人扩大到 4 或 5 名，以减少积压。之后它将再次缩减，因为 CPU 利用率对于 4 或 5 个工人来说太低了。
正是在这一点上，所有工作人员的 CPU 利用率下降到接近 0%，整个管道开始大幅落后。结果是工作人员的数量再次扩大到更高的数量，并且管道进一步处理数据。再次减少积压后，工人数量逐渐减少，出现同样的问题。
metrics
我们注意到，在 GroupByKey 步骤中，输入吞吐量大致保持不变，但输出吞吐量下降到 0。
GroupByKey throughput
我知道使用 GroupByKey 可以有热键，但我希望 1 个工人的 CPU 利用率非常高，而其他工人则无事可做。
有谁知道可能导致此问题的原因吗？
问题是由于会话 window 与 groupbykey 的组合使用、pubsub 无界源的水印如何工作以及确认发送到 pubsub 的时间引起的。
我们的会话 window 有 10 秒的间隔，有时几分钟内没有输出任何消息（由于没有配置早期触发器并且消息在 10 秒内连续到达相同的键会话间隙）。因为这些步骤是我们管道实际执行中第一个融合阶段的一部分，这导致一些消息没有被 pubsub 确认（只有当第一个融合阶段完成时才会发送 ack）。订阅中最老的未确认消息时间不断增加，导致水印无法推进。
由于确认截止日期设置为 10 分钟，此问题变得更加直言不讳。当工人数量减少时，这导致了原始问题中描述的问题。
我们能够通过在创建会话 window（使用 groupbykey）之前添加 Reshuffle 并缩短确认截止日期来解决这个问题。
https://cloud.google.com/blog/products/data-analytics/handling-duplicate-data-in-streaming-pipeline-using-pubsub-dataflow
https://cloud.google.com/dataflow/docs/guides/deploying-a-pipeline#fusion-optimization
在我们的流媒体管道中，我们从 pubsub 读取数据，进行一些验证，然后在 10 秒间隔会话 window 中按键对其进行分组。之后数据被进一步处理并再次写入bigtable和pubsub。
我们使用的是 apache beam 2.28 和数据流流引擎。白天我们处理的数据比晚上多，管道会自动增加工作人员 (n2d-standard-4) 的数量。大多数情况下，它会从 2 名工人扩大到 4 或 5 名，以减少积压。之后它将再次缩减，因为 CPU 利用率对于 4 或 5 个工人来说太低了。
正是在这一点上，所有工作人员的 CPU 利用率下降到接近 0%，整个管道开始大幅落后。结果是工作人员的数量再次扩大到更高的数量，并且管道进一步处理数据。再次减少积压后，工人数量逐渐减少，出现同样的问题。
metrics
我们注意到，在 GroupByKey 步骤中，输入吞吐量大致保持不变，但输出吞吐量下降到 0。
GroupByKey throughput
我知道使用 GroupByKey 可以有热键，但我希望 1 个工人的 CPU 利用率非常高，而其他工人则无事可做。
有谁知道可能导致此问题的原因吗？
问题是由于会话 window 与 groupbykey 的组合使用、pubsub 无界源的水印如何工作以及确认发送到 pubsub 的时间引起的。
我们的会话 window 有 10 秒的间隔，有时几分钟内没有输出任何消息（由于没有配置早期触发器并且消息在 10 秒内连续到达相同的键会话间隙）。因为这些步骤是我们管道实际执行中第一个融合阶段的一部分，这导致一些消息没有被 pubsub 确认（只有当第一个融合阶段完成时才会发送 ack）。订阅中最老的未确认消息时间不断增加，导致水印无法推进。
由于确认截止日期设置为 10 分钟，此问题变得更加直言不讳。当工人数量减少时，这导致了原始问题中描述的问题。
我们能够通过在创建会话 window（使用 groupbykey）之前添加 Reshuffle 并缩短确认截止日期来解决这个问题。
https://cloud.google.com/blog/products/data-analytics/handling-duplicate-data-in-streaming-pipeline-using-pubsub-dataflow https://cloud.google.com/dataflow/docs/guides/deploying-a-pipeline#fusion-optimization