Template MongoDB ke BigQuery

Template ini membuat pipeline batch yang membaca dokumen dari MongoDB dan menulisnya ke BigQuery.

Jika ingin mengambil data aliran data perubahan MongoDB, Anda dapat menggunakan template MongoDB ke BigQuery (CDC).

Persyaratan pipeline

Set data BigQuery target harus ada.
Instance MongoDB sumber harus dapat diakses dari mesin pekerja Dataflow.

Format output

Format data output bergantung pada nilai parameter userOption. Jika userOption adalah NONE, outputnya akan memiliki skema berikut. Kolom source_data berisi dokumen dalam format JSON.

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

Jika userOption adalah FLATTEN, pipeline akan meratakan dokumen dan menulis kolom tingkat atas sebagai kolom tabel. Misalnya, dokumen dalam koleksi MongoDB berisi kolom berikut:

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

Dengan menggunakan FLATTEN, output-nya memiliki skema berikut. Kolom timestamp ditambahkan oleh template.

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

Parameter template

Parameter	Deskripsi
`mongoDbUri`	URI koneksi MongoDB dalam format `mongodb+srv://:@`.
`database`	Database di MongoDB yang akan digunakan untuk membaca koleksi. Misalnya: `my-db`.
`collection`	Nama koleksi di dalam database MongoDB. Misalnya: `my-collection`.
`outputTableSpec`	Tabel BigQuery yang menjadi tujuan penulisan. Misalnya, `bigquery-project:dataset.output_table`.
`userOption`	`FLATTEN` atau `NONE`. `FLATTEN` meratakan dokumen ke level pertama. `NONE` menyimpan seluruh dokumen sebagai string JSON.
`javascriptDocumentTransformGcsPath`	(Opsional) URI Cloud Storage dari file `.js` yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan. Misalnya, `gs://my-bucket/my-udfs/my_file.js`.
`javascriptDocumentTransformFunctionName`	(Opsional) Nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan. Misalnya, jika kode fungsi JavaScript Anda adalah `myTransform(inJson) { /...do stuff.../ }`, nama fungsi adalah `myTransform`. Untuk contoh UDF JavaScript, lihat Contoh UDF.
`useStorageWriteApi`	(Opsional) Jika `true`, pipeline akan menggunakan BigQuery Storage Write API. Nilai defaultnya adalah `false`. Untuk informasi selengkapnya, lihat Menggunakan Storage Write API.
`useStorageWriteApiAtLeastOnce`	(Opsional) Saat menggunakan Storage Write API, menentukan semantik penulisan. Untuk menggunakan semantik minimal satu kali , tetapkan parameter ini ke `true`. Untuk menggunakan semantik tepat satu kali, tetapkan parameter ke `false`. Parameter ini hanya berlaku jika `useStorageWriteApi` adalah `true`. Nilai defaultnya adalah `false`.

Fungsi yang ditentukan pengguna

Jika ingin, Anda dapat memperluas template ini dengan menulis fungsi yang ditentukan pengguna (UDF) di JavaScript. Template memanggil UDF untuk setiap elemen input. Payload elemen diserialisasi sebagai string JSON.

Untuk menggunakan UDF, upload file JavaScript ke Cloud Storage dan tetapkan parameter template berikut:

Parameter	Deskripsi
`javascriptDocumentTransformGcsPath`	Lokasi Cloud Storage file JavaScript.
`javascriptDocumentTransformFunctionName`	Nama fungsi JavaScript.

Untuk mengetahui informasi selengkapnya, lihat Membuat fungsi yang ditentukan pengguna untuk template Dataflow.

Spesifikasi fungsi

UDF memiliki spesifikasi berikut:

Input: dokumen MongoDB.

Output: objek yang diserialisasi sebagai string JSON. Jika userOption adalah NONE, objek JSON harus menyertakan properti bernama _id yang berisi ID dokumen.

Menjalankan template

Konsol

Buka halaman Create job from template Dataflow.

Buka Buat tugas dari template

Di kolom Job name, masukkan nama pekerjaan yang unik.
Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region default-nya adalah us-central1.
Untuk daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.
Dari menu drop-down Dataflow template, pilih the MongoDB to BigQuery template.
Di kolom parameter yang disediakan, masukkan parameter value Anda.
Klik Run job.

gcloud

Di shell atau terminal Anda, jalankan template:

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

Ganti kode berikut:

PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
JOB_NAME: nama pekerjaan unik pilihan Anda
REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
VERSION: versi template yang ingin Anda gunakan
Anda dapat menggunakan nilai berikut:
- latest untuk menggunakan versi terbaru template, yang tersedia di folder induk tidak bertanggal di bucket— gs://dataflow-templates-REGION_NAME/latest/
- nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat di folder induk bertanggal masing-masing dalam bucket— gs://dataflow-templates-REGION_NAME/
Perhatian: Versi terbaru template mungkin diperbarui dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak memengaruhi alur kerja produksi Anda.
OUTPUT_TABLE_SPEC: nama tabel BigQuery target Anda.
MONGO_DB_URI: URI MongoDB Anda.
DATABASE: database MongoDB Anda.
COLLECTION: koleksi MongoDB Anda.
USER_OPTION: FLATTEN atau NONE.

API

Untuk menjalankan template menggunakan REST API, kirim permintaan HTTP POST. Untuk informasi selengkapnya tentang API dan cakupan otorisasinya, lihat 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-LOCATION/VERSION/flex/MongoDB_to_BigQuery",
   }
}

Ganti kode berikut:

PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
JOB_NAME: nama pekerjaan unik pilihan Anda
LOCATION: region tempat Anda ingin men-deploy tugas Dataflow, misalnya us-central1
VERSION: versi template yang ingin Anda gunakan
Anda dapat menggunakan nilai berikut:
- latest untuk menggunakan versi terbaru template, yang tersedia di folder induk tidak bertanggal di bucket— gs://dataflow-templates-REGION_NAME/latest/
- nama versi, seperti 2023-09-12-00_RC00, untuk menggunakan versi template tertentu, yang dapat ditemukan bertingkat di folder induk bertanggal masing-masing dalam bucket— gs://dataflow-templates-REGION_NAME/
Perhatian: Versi terbaru template mungkin diperbarui dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru agar perubahan yang dapat menyebabkan gangguan ini tidak memengaruhi alur kerja produksi Anda.
OUTPUT_TABLE_SPEC: nama tabel BigQuery target Anda.
MONGO_DB_URI: URI MongoDB Anda.
DATABASE: database MongoDB Anda.
COLLECTION: koleksi MongoDB Anda.
USER_OPTION: FLATTEN atau NONE.

Kode sumber template

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.KMSUtils.maybeDecrypt;

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.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 java.io.IOException;
import javax.script.ScriptException;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO;
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.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 = "mongodb-to-bigquery",
    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,
          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.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",
            MongoDbIO.read()
                .withUri(mongoDbUri)
                .withDatabase(options.getDatabase())
                .withCollection(options.getCollection()))
        .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;
  }
}

Langkah selanjutnya

Pelajari Template Dataflow.
Lihat daftar template yang disediakan Google.