有没有办法在 GCP 中对批处理数据流作业进行排队？答案

【问题标题】：Is there a way to queue Batch Dataflow jobs in GCP?有没有办法在 GCP 中对批处理数据流作业进行排队？
【发布时间】：2021-08-28 02:29:46
【问题描述】：

长话短说，我有一个 cron 作业，每天在指定时间将一堆文件上传到 Cloud Storage 存储桶中。所有这些存储桶都有一个关联的 Pub/Sub 通知主题，该主题在文件创建事件上触发。每个事件都会触发一个 Dataflow 作业来处理该文件。

问题是这会在几秒钟内实例化 100 个并行批处理作业。每项工作都会使用 HTTP 请求来抨击我的下游服务。服务无法足够快地扩展并开始引发连接被拒绝错误。

为了限制这些请求，我限制了每个 Dataflow 作业的可用工作人员数量。此外，为我的下游服务增加了资源，并将其 targetCPUUtilizationPercentage 降低到 50%，以便他们有时间扩大规模。所有这些都将错误数量从 4000 万个失败请求大幅减少到约 50000 个。

但是，我想知道是否有办法将这些批处理作业排队？只有 5 到 10 个作业并行运行将为我的下游服务提供更多喘息空间。

FlexRS 是我尚未尝试的东西，但我认为它不会有太大帮助，因为该算法针对 COST 或 SPEED 进行了优化。这两者都不是问题。

注意：我公司的所有基础架构都是基于 GCP 的。随意提出任何其他非基于队列的建议/优化。

【问题讨论】：

每个作业是否在不同的文件上执行相同的事情（相同的管道）？文件的大小是多少？
是的。工作是一样的。主要是逐行读取文件（JSON 或 CSV）。文件大小范围从 20 MB 到最大 4 GB。
我认为您可以通过编程方式创建管道并一一运行。我假设您有一个读取 Pub/Sub 以启动其他 Dataflow 作业的流式作业。如果是这种情况，则不要使用 Dataflow 从 Pub/Sub 主题中读取数据，而是编写一个程序来订阅该主题并将 Dataflow 管道排入队列。将 Dataflow 管道排入队列以在当前管道停止时执行。您可以随时检查管道结果的状态。

【解决方案1】：

在以前的公司，我也遇到过同样的问题。我们通过流式处理解决了这个问题：

可以解决您的问题

【讨论】：

我们决定使用批处理，因为我们预计将来会有 50 个甚至 100 个 gig 文件。即使可能，将这些下载到数据流中的内存也可能非常昂贵。你怎么看？
下载，逐行读取创建你的PCollection，删除文件。 FileIO 做同样的事情！在同一个转换中执行操作很重要，因为如果您更改步骤，它可能会在另一个工作人员上执行，您可能会丢失上下文

【解决方案2】：

您在架构中已经有一个 pub/sub 您可以创建一个主题并像使用它一样使用它

...在文件创建事件上触发的发布/订阅通知主题。每个事件都会向主题 job_queue 发布一条消息。模板用于从中读取并触发 Dataflow 作业来处理该文件，详细信息在上一步中发送的 msg 中

【讨论】：