自动将数据上传到 Google Cloud Storage 和 BigQuery

Question

我正在尝试将一些数据从我的应用程序传输到 Big Query，以便我可以在其上运行 SQL 查询。以下是我正在采取的步骤，并想知道是否有工具可以自动执行此过程。

1. 从终端手动下载应用程序数据，格式为 JSON。

2. 运行 python 脚本将此文件解析为可读的 JSON 并进行修改（例如，将空格替换为下划线等，以便能够上传到 Google Cloud Storage (GCS)。

3. 手动将此更新后的文件上传到 GCS。

4. 使用来自 GCS 的这些数据，使用架构在 Big Query 中手动生成数据集和表。

有没有办法使用云工具自动化整个过程？我查看了谷歌数据流，但它似乎不是这个工具。或者我应该只拥有一个可以完成上述所有操作的 python 程序？

请指教。

Answer 1

为了解决您的问题，我将编写 3 个服务（例如 Cloud Functions 或 Cloud Run。它是无服务器的，按使用付费，并且通过免费层级，您最终将无需支付任何费用）。

• 服务 1：下载文件并将其作为原始文件存储在云存储中
• 服务 2：从存储中获取文件，转换文件并将其作为干净文件存储回存储中
• 服务 3：在 BigQuery 中加载数据。

这里是事件驱动架构中的流程：

• 创建一个云调度程序，在您需要下载文件时调用您的服务 1（每天一次，每小时一次，...）
• 当原始文件存储在 Cloud Storage 中时，会发出一个事件。抓住它来触发你的服务 2（有两种方式可以实现，如果你不知道怎么做，请告诉我）
• 当干净的文件存储在 Cloud Storage 中时，会发出一个事件。抓住它来触发你的 Service 3

您也可以使用 Cloud Workflow 通过编排来实现相同的目标

• 在 Cloud Workflow 上部署您的工作流
• 创建一个云计划程序，在您需要下载文件时调用您的工作流（每天一次，每小时一次，...）
• 让工作流运行：
  • 调用服务 1。
  • 服务 1 结束后，调用服务 2
  • 服务 2 结束后，调用服务 3

---

首先，您可以在同一个服务中执行 3 个服务的所有操作。 POC 更容易。但是为了解决方案的可扩展性和演进，最好的设计是限制每个服务的责任。

---

编辑 1

要从 Cloud Storage 捕捉事件，您有多种解决方案

• 如果您使用 Cloud Functions，您可以plug it directly on GCS event。 它被命名为后台函数
• 如果您使用http Cloud Functions或Cloud Run，您可以publish the Cloud Storage events in PubSub。然后，在 PubSub 上，您可以创建一个 push subscription to call an HTTP endpoint（http Cloud Functions 或 Cloud Run）。注意这里的安全。如果你被卡住了，再问我一次。 这是我的首选和最通用的解决方案
• 如果您使用Cloud Run，您也可以使用Eventarc来捕获Storage事件并调用Cloud Run服务。 EventArc 将之前的解决方案封装在“一键部署”中

Answer 2

了解apache气流和这个框架的谷歌云版本：composer。您可以设置每个任务的工作流程是您提到的步骤。 Airflow 支持很多运算符：BashOperator、PythonOperator、BigQuery、...