用于不同架构的 Google Dataflow 管道

Question

我有一个产品来定义和配置业务工作流程。该产品的一部分是一个表单构建器，它使用户能够设置不同的表单。

整个表单数据以以下结构支持在 MongoDB 上

- form_schemas

{
  "_id" : "",
  "name" : "",
  "account_id" : "",
  "fields" : [
    {
      "label" : "Enter Email",
      "name" : "email",
      "type" : "text",
      "required" : "true",
      "hidden" : "false",
      "additional_config" : { }
    },
    {
      "label" : "Select DOB",
      "name" : "dob",
      "type" : "date",
      "required" : "true",
      "hidden" : "false",
      "additional_config" : { }
    }
    ...
  ]
}

- form_datas

{
  "workflow_template_id" : "",
  "account_id" : ""
  "data" : {
    "email" : "xyx@gmail.com",
    "dob" : "2001-04-05"
  },
  "created_at" : "",
  "updated_at" : ""

}

如上所示，该表格可用于各种不同的业务。 但是，我正在研究数据管道以定期将数据传输到 Google Bigquery 进行分析。

在 BQ 方面，我为每个工作流维护单独的表

我目前有一个完全使用 Google Cloud Functions 编写的有效解决方案。 我有一个定期运行的 Google 调度程序作业，调用不同的云功能。 云函数在高层做以下事情

• 为每个架构迭代
• 读取自上次运行以来每个模式的数据 mongodb（作为游标）
• 对于每一行数据，运行自定义转换逻辑（这包括转换各种嵌套数据类型，如网格/查找等）
• 将每一行转换后的数据直接作为流作为 ndjson 写入 Google Cloud Storage 上

我上面的解决方案为我提供了，

• 完全控制转换
• 简单部署

但是，由于它全部在 CF 上，我受到每次跑步 9 分钟的限制。 这本质上提出了很多分页要求，特别是如果需要从一开始就重新生成完整的数据

虽然上述解决方案目前运行良好，但我正在研究其他无服务器选项，例如 Google 数据流。由于我刚刚开始研究数据流/apache Beam，所以我想知道

如果我要在梁上写一个管道，我应该采用同样的方法

1. 提取（逐行）-> 转换-> 加载（GCS）-> 加载（BQ）

或

1. 提取（整个数据为 JSON）-> 加载到 GCS -> 转换（梁）-> 加载到 GCS -> 加载到 BQ

如果有更好的选择来处理整个数据，请告诉我。

Answer 1

通常，this sort of process writes raw data to GCS and then transforms 进入 Bigquery。这样做是为了当您发现转换中的缺陷（这是不可避免的）和需求变化（也是不可避免的）时，您可以使用新代码重放数据。

理想情况下，转换之前的步骤由Change Data Capture (CDC) 工具自动完成。有很多 CDC 工具，但 Debezium 正在接管，因为它可靠且免费。有一个Debezium connector to get data from MongoDB 和examples of how to put Debezium CDC into Bigquery。

如果您要编写将数据放入 GCS 的代码，我建议您考虑使用 Apache Parquet 而不是 NDJSON 作为格式。性能和成本会更好，而且我发现具有数据类型的格式更易于使用。