MongoDB 到 BigQuery 範本

Beta 版

這項功能適用《服務專屬條款》中「一般服務條款」一節的《正式發布前產品條款》。正式發布前功能是依「原樣」提供，支援服務可能受限。 詳情請參閱推出階段說明。

這個範本會建立批次管道，從 MongoDB 讀取文件並寫入 BigQuery。

如要擷取 MongoDB 變更串流資料，可以使用 MongoDB 至 BigQuery (CDC) 範本。

管道相關規定

• 目標 BigQuery 資料集必須已存在。
• 來源 MongoDB 執行個體必須可從 Dataflow worker 機器存取。

輸出格式

輸出記錄的格式取決於 userOption 參數的值。如果 userOption 是 NONE，輸出內容會採用下列結構定義。source_data 欄位包含 JSON 格式的文件。

  [
    {"name":"id","type":"STRING"},
    {"name":"source_data","type":"STRING"},
    {"name":"timestamp","type":"TIMESTAMP"}
  ]
  

如果 userOption 為 FLATTEN，管道會將文件扁平化，並將頂層欄位寫入為資料表資料欄。舉例來說，假設 MongoDB 集合中的文件包含下列欄位：

• "_id" (string)
• "title" (string)
• "genre" (string)

使用 FLATTEN 時，輸出內容會採用下列結構定義。timestamp 欄位是由範本新增。

  [
    {"name":"_id","type":"STRING"},
    {"name":"title","type":"STRING"},
    {"name":"genre","type":"STRING"},
    {"name":"timestamp","type":"TIMESTAMP"}
  ]
  

如果 userOption 為 JSON，管道會以 BigQuery JSON 格式儲存文件。BigQuery 內建支援 JSON 資料，使用 JSON 資料類型即可。詳情請參閱「在 GoogleSQL 中使用 JSON 資料」。

範本參數

必要參數

• mongoDbUri：MongoDB 連線 URI，格式為 mongodb+srv://:@.。
• 資料庫：MongoDB 中用於讀取集合的資料庫。例如：my-db。
• collection：MongoDB 資料庫中的集合名稱。例如：my-collection。
• userOption：FLATTEN、JSON 或 NONE。FLATTEN 會將文件扁平化為單一層級。JSON 會以 BigQuery JSON 格式儲存文件。NONE 會將整份文件儲存為 JSON 格式的字串。預設值為：NONE。
• outputTableSpec：要寫入的 BigQuery 資料表。例如：bigquery-project:dataset.output_table。

選用參數

• KMSEncryptionKey：Cloud KMS 加密金鑰，用於解密 mongodb uri 連線字串。如果傳入 Cloud KMS 金鑰，mongodb uri 連線字串必須全部以加密形式傳入。例如：projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key。
• filter：JSON 格式的 Bson 篩選器。例如：{ "val": { $gt: 0, $lt: 9 }}。
• useStorageWriteApi：如果為 true，管道會使用 BigQuery Storage Write API (https://cloud.google.com/bigquery/docs/write-api)。預設值為 false。詳情請參閱「使用 Storage Write API」(https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api)。
• useStorageWriteApiAtLeastOnce：使用 Storage Write API 時，指定寫入語意。如要使用至少一次語意 (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics)，請將這個參數設為 true。如要使用僅一次語意，請將參數設為 false。只有在 useStorageWriteApi 為 true 時，這項參數才會生效。預設值為 false。
• bucketAuto：如果為 true，系統會使用 $bucketAuto 匯總階段，依據 _id 將集合分割成分區，因此讀取作業會在多個工作站上平行執行。與 splitVector 不同，$bucketAuto 也允許在 MongoDB Atlas 上使用。分區數量由 numSplits 控制。如果集合的 _id 值不是 ObjectId，則必須使用 Apache Beam 2.77.0 以上版本。預設為 false。
• numSplits：讀取作業要分割成的分區數量。當 bucketAuto 為 true 時，0 會使用 10 個分區。當 bucketAuto 為 false 時，如果值大於 0，系統會使用 splitVector 指令分割讀取作業，但 MongoDB Atlas 不允許這麼做，且 0 會將集合讀取為單一分區。預設值為 0。
• bigQuerySchemaPath：BigQuery JSON 結構定義的 Cloud Storage 路徑。例如：gs://your-bucket/your-schema.json。
• javascriptDocumentTransformGcsPath：定義要使用的 JavaScript user-defined function (UDF) 的 .js 檔案 Cloud Storage URI。例如：gs://your-bucket/your-transforms/*.js。
• javascriptDocumentTransformFunctionName：要使用的 JavaScript user-defined function (UDF) 名稱。例如，如果您的 JavaScript 函式程式碼是 myTransform(inJson) { /*...do stuff...*/ }，則函式名稱就是 myTransform。如需 JavaScript UDF 範例，請參閱 UDF 範例 (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples)。例如：transform。

User-defined function

您可以視需要使用 JavaScript 撰寫 user-defined function (UDF)，擴充這個範本。範本會針對每個輸入元素呼叫 UDF。 元素酬載會序列化為 JSON 字串。

如要使用 UDF，請將 JavaScript 檔案上傳至 Cloud Storage，並設定下列範本參數：

參數	說明
javascriptDocumentTransformGcsPath	JavaScript 檔案的 Cloud Storage 位置。
javascriptDocumentTransformFunctionName	JavaScript 函式名稱。

詳情請參閱「為 Dataflow 範本建立 user-defined function」。

函式規格

UDF 的規格如下：

輸入：MongoDB 文件。

輸出：以 JSON 字串序列化的物件。如果 userOption 為 NONE，JSON 物件必須包含名為 _id 的屬性，其中包含文件 ID。

執行範本

控制台

1. 前往 Dataflow 的「Create job from template」(依據範本建立工作) 頁面。
前往「依範本建立工作」
2. 在「Job name」(工作名稱) 欄位中，輸入專屬工作名稱。
3. 選用：如要使用區域端點，請從下拉式選單中選取值。預設區域為 us-central1。

   如需可執行 Dataflow 工作的區域清單，請參閱「Dataflow 位置」。

4. 從「Dataflow template」(Dataflow 範本) 下拉式選單中，選取「MongoDB to BigQuery」(MongoDB 到 BigQuery) 範本。
5. 在提供的參數欄位中輸入參數值。
6. 按一下「Run Job」(執行工作)。

gcloud

注意：如要使用 the Google Cloud CLI 執行 flex範本，您必須安裝 Google Cloud CLI 284.0.0 以上版本。

在殼層或終端機中執行範本：

gcloud dataflow flex-template run JOB_NAME \
    --project=PROJECT_ID \
    --region=REGION_NAME \
    --template-file-gcs-location=gs://dataflow-templates/VERSION/flex/MongoDB_to_BigQuery \
    --parameters \
outputTableSpec=OUTPUT_TABLE_SPEC,\
mongoDbUri=MONGO_DB_URI,\
database=DATABASE,\
collection=COLLECTION,\
userOption=USER_OPTION

更改下列內容：

• PROJECT_ID：您要執行 Dataflow 工作的 Google Cloud 專案 ID
• JOB_NAME：您選擇的不重複工作名稱
• REGION_NAME：您要部署 Dataflow 工作的區域，例如 us-central1
• VERSION：您要使用的範本版本

  您可以使用下列值：

  • latest，使用範本的最新版本，該版本位於 bucket 中非依日期命名的上層資料夾：gs://dataflow-templates/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在 bucket 中相應的依日期命名的上層資料夾中：gs://dataflow-templates/
  注意：範本的最新版本可能會隨著破壞性變更而有所更新。您的正式環境使用的範本應該來自最近依日期命名的上層資料夾，以免這些破壞性變更影響正式環境流程。
• OUTPUT_TABLE_SPEC：目標 BigQuery 資料表名稱。
• MONGO_DB_URI：您的 MongoDB URI。
• DATABASE：您的 MongoDB 資料庫。
• COLLECTION：您的 MongoDB 集合。
• USER_OPTION：FLATTEN、JSON 或 NONE。

API

如要使用 REST API 執行範本，請傳送 HTTP POST 要求。如要進一步瞭解 API 和授權範圍，請參閱 projects.templates.launch。

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
          "inputTableSpec": "INPUT_TABLE_SPEC",
          "mongoDbUri": "MONGO_DB_URI",
          "database": "DATABASE",
          "collection": "COLLECTION",
          "userOption": "USER_OPTION"
      },
      "containerSpecGcsPath": "gs://dataflow-templates/VERSION/flex/MongoDB_to_BigQuery",
   }
}

更改下列內容：

• PROJECT_ID：您要執行 Dataflow 工作的 Google Cloud 專案 ID
• JOB_NAME：您選擇的不重複工作名稱
• LOCATION：您要部署 Dataflow 工作的區域，例如 us-central1
• VERSION：您要使用的範本版本

  您可以使用下列值：

  • latest，使用範本的最新版本，該版本位於 bucket 中非依日期命名的上層資料夾：gs://dataflow-templates/latest/
  • 版本名稱，例如 2023-09-12-00_RC00，可使用特定版本的範本，該範本會以巢狀結構存放在 bucket 中相應的依日期命名的上層資料夾中：gs://dataflow-templates/
  注意：範本的最新版本可能會隨著破壞性變更而有所更新。您的正式環境使用的範本應該來自最近依日期命名的上層資料夾，以免這些破壞性變更影響正式環境流程。
• OUTPUT_TABLE_SPEC：目標 BigQuery 資料表名稱。
• MONGO_DB_URI：您的 MongoDB URI。
• DATABASE：您的 MongoDB 資料庫。
• COLLECTION：您的 MongoDB 集合。
• USER_OPTION：FLATTEN、JSON 或 NONE。

範本原始碼

Java

/*
 * Copyright (C) 2019 Google LLC
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package com.google.cloud.teleport.v2.mongodb.templates;

import static com.google.cloud.teleport.v2.utils.GCSUtils.getGcsFileAsString;
import static com.google.cloud.teleport.v2.utils.KMSUtils.maybeDecrypt;

import com.google.api.client.json.gson.GsonFactory;
import com.google.api.services.bigquery.model.TableRow;
import com.google.api.services.bigquery.model.TableSchema;
import com.google.cloud.teleport.metadata.Template;
import com.google.cloud.teleport.metadata.TemplateCategory;
import com.google.cloud.teleport.v2.common.UncaughtExceptionLogger;
import com.google.cloud.teleport.v2.mongodb.options.MongoDbToBigQueryOptions.BigQueryWriteOptions;
import com.google.cloud.teleport.v2.mongodb.options.MongoDbToBigQueryOptions.JavascriptDocumentTransformerOptions;
import com.google.cloud.teleport.v2.mongodb.options.MongoDbToBigQueryOptions.MongoDbOptions;
import com.google.cloud.teleport.v2.mongodb.options.MongoDbToBigQueryOptions.MongoDbReadOptions;
import com.google.cloud.teleport.v2.mongodb.templates.MongoDbToBigQuery.Options;
import com.google.cloud.teleport.v2.options.BigQueryStorageApiBatchOptions;
import com.google.cloud.teleport.v2.transforms.JavascriptDocumentTransformer.TransformDocumentViaJavascript;
import com.google.cloud.teleport.v2.utils.BigQueryIOUtils;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Strings;
import java.io.IOException;
import javax.script.ScriptException;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
import org.apache.beam.sdk.io.mongodb.FindQuery;
import org.apache.beam.sdk.io.mongodb.MongoDbIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.bson.BsonDocument;
import org.bson.Document;

/**
 * The {@link MongoDbToBigQuery} pipeline is a batch pipeline which ingests data from MongoDB and
 * outputs the resulting records to BigQuery.
 *
 * <p>Check out <a
 * href="https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/v2/mongodb-to-googlecloud/README_MongoDB_to_BigQuery.md">README</a>
 * for instructions on how to use or modify this template.
 */
@Template(
    name = "MongoDB_to_BigQuery",
    category = TemplateCategory.BATCH,
    displayName = "MongoDB to BigQuery",
    description =
        "The MongoDB to BigQuery template is a batch pipeline that reads documents from MongoDB and writes them to "
            + "BigQuery as specified by the <code>userOption</code> parameter.",
    optionsClass = Options.class,
    flexContainerName = "googlecloud-and-mongodb",
    documentation =
        "https://cloud.google.com/dataflow/docs/guides/templates/provided/mongodb-to-bigquery",
    contactInformation = "https://cloud.google.com/support",
    preview = true,
    requirements = {
      "The target BigQuery dataset must exist.",
      "The source MongoDB instance must be accessible from the Dataflow worker machines."
    })
public class MongoDbToBigQuery {
  /**
   * Options supported by {@link MongoDbToBigQuery}
   *
   * <p>Inherits standard configuration options.
   */
  public interface Options
      extends PipelineOptions,
          MongoDbOptions,
          MongoDbReadOptions,
          BigQueryWriteOptions,
          BigQueryStorageApiBatchOptions,
          JavascriptDocumentTransformerOptions {}

  private static class ParseAsDocumentsFn extends DoFn<String, Document> {
    @ProcessElement
    public void processElement(ProcessContext context) {
      context.output(Document.parse(context.element()));
    }
  }

  public static void main(String[] args)
      throws ScriptException, IOException, NoSuchMethodException {
    UncaughtExceptionLogger.register();

    Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);

    BigQueryIOUtils.validateBQStorageApiOptionsBatch(options);

    run(options);
  }

  public static boolean run(Options options)
      throws ScriptException, IOException, NoSuchMethodException {
    Pipeline pipeline = Pipeline.create(options);
    String userOption = options.getUserOption();

    TableSchema bigquerySchema;

    // Get MongoDbUri plain text or base64 encrypted with a specific KMS encryption key
    String mongoDbUri = maybeDecrypt(options.getMongoDbUri(), options.getKMSEncryptionKey()).get();

    if (options.getBigQuerySchemaPath() != null) {
      // initialize FileSystem to read from GCS
      FileSystems.setDefaultPipelineOptions(options);
      String jsonSchema = getGcsFileAsString(options.getBigQuerySchemaPath());
      GsonFactory gf = new GsonFactory();
      bigquerySchema = gf.fromString(jsonSchema, TableSchema.class);
    } else if (options.getJavascriptDocumentTransformFunctionName() != null
        && options.getJavascriptDocumentTransformGcsPath() != null) {
      bigquerySchema =
          MongoDbUtils.getTableFieldSchemaForUDF(
              mongoDbUri,
              options.getDatabase(),
              options.getCollection(),
              options.getJavascriptDocumentTransformGcsPath(),
              options.getJavascriptDocumentTransformFunctionName(),
              options.getUserOption());
    } else {
      bigquerySchema =
          MongoDbUtils.getTableFieldSchema(
              mongoDbUri, options.getDatabase(), options.getCollection(), options.getUserOption());
    }

    pipeline
        .apply("Read Documents", buildReadDocuments(options, mongoDbUri))
        .apply(
            "UDF",
            TransformDocumentViaJavascript.newBuilder()
                .setFileSystemPath(options.getJavascriptDocumentTransformGcsPath())
                .setFunctionName(options.getJavascriptDocumentTransformFunctionName())
                .build())
        .apply(
            "Transform to TableRow",
            ParDo.of(
                new DoFn<Document, TableRow>() {

                  @ProcessElement
                  public void process(ProcessContext c) {
                    Document document = c.element();
                    TableRow row = MongoDbUtils.getTableSchema(document, userOption);
                    c.output(row);
                  }
                }))
        .apply(
            "Write to Bigquery",
            BigQueryIO.writeTableRows()
                .to(options.getOutputTableSpec())
                .withSchema(bigquerySchema)
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));
    pipeline.run();
    return true;
  }

  @VisibleForTesting
  static MongoDbIO.Read buildReadDocuments(Options options, String mongoDbUri) {
    MongoDbIO.Read readDocuments =
        MongoDbIO.read()
            .withUri(mongoDbUri)
            .withDatabase(options.getDatabase())
            .withCollection(options.getCollection())
            .withBucketAuto(options.getBucketAuto())
            .withNumSplits(options.getNumSplits());

    String filterJson = options.getFilter();
    BsonDocument filter;
    if (!Strings.isNullOrEmpty(filterJson)
        && !(filter = BsonDocument.parse(filterJson)).isEmpty()) {
      readDocuments = readDocuments.withQueryFn(FindQuery.create().withFilters(filter));
    }
    return readDocuments;
  }
}

後續步驟

• 瞭解 Dataflow 範本。
• 請參閱 Google 提供的範本清單。