如何使用 Dataflow 批量处理从 Google Pub/Sub 到 Cloud Storage 的数据？

Question

我正在构建一个更改数据捕获管道，它从 MYSQL 数据库读取数据并在 BigQuery 中创建一个副本。我将推送 Pub/Sub 中的更改并使用 Dataflow 将它们传输到 Google Cloud Storage。我已经能够弄清楚如何流式传输更改，但我需要对我的数据库中的一些表运行批处理。

在从 Pub/Sub 等无限源读取数据时，能否使用 Dataflow 运行批处理作业？我可以运行此批处理作业将数据从 Pub/Sub 传输到 Cloud Storage，然后将此数据加载到 BigQuery 吗？我想要批处理作业，因为流作业成本更高。

Answer 1

感谢您的精确。

首先，当您在 Dataflow（Beam 框架）中使用 PubSub 时，只能在 streaming mode 中使用

Cloud Pub/Sub 源和接收器目前仅在远程执行期间的流式管道中受支持。

如果您的流程不需要实时，您可以跳过 Dataflow 并节省资金。您可以将 Cloud Functions 或 Cloud Run 用于我建议您的流程（如果您愿意，也可以使用 App Engine，但不是我的第一个建议）。

在这两种情况下，创建一个由Cloud Scheduler 定期（每周？）触发的流程（Cloud Run 或 Cloud Function）。

解决方案 1

• 将您的流程连接到拉取订阅
• 每次您读取一条消息（或一条消息，例如 1000 条）时，将流写入 BigQuery。 -> 但是，大查询上的流写入不是免费的（每 Gb 0.05 美元）
• 循环直到队列为空。将超时设置为最大值（使用 Cloud Function 为 9 分钟，使用 Cloud Run 为 15 分钟）以防止出现任何超时问题

解决方案 2

• 将您的流程连接到拉取订阅
• 读取大量消息（例如 1000 条）并将它们写入内存（写入数组）。
• 循环直到队列为空。将超时设置为最大值（使用 Cloud Function 为 9 分钟，使用 Cloud Run 为 15 分钟）以防止出现任何超时问题。还将内存设置为最大值 (2Gb) 以防止内存不足崩溃。
• 从您的内存数据数组创建加载作业到 BigQuery。 -> 此处的加载作业是免费的，每天每张桌子的加载作业限制为 1000 个。

但是，如果您的应用 + 数据大小大于 ma 内存值，则此解决方案可能会失败。另一种方法是在 GCS 中创建一个文件，例如，每 100 万行（取决于每行的大小和内存占用）。使用唯一前缀命名文件，例如日期 (YYYYMMDD-tempFileXX)，并在每次创建文件时递增 XX。然后，创建一个加载作业，不是从内存中的数据，而是使用 GCS 中的数据，文件名中带有通配符 (gs://myBucket/YYYYMMDD-tempFile*)。这样所有与前缀匹配的文件都会被加载。

建议 PubSub 消息在 pubsub 订阅中最多保留 7 天。我建议您至少每 3 天触发一次流程，以便在将消息删除到订阅中之前有时间做出反应和调试。

个人经验 写入 BigQuery 的流对于少量数据而言成本低廉。对于一些美分，我建议您考虑第一个解决方案是您可以为此付费。管理和代码更小/更容易！