Template MongoDB ke BigQuery

Beta

Fitur ini tunduk pada "Persyaratan Penawaran Pra-GA" di bagian Persyaratan Layanan Umum dalam Persyaratan Khusus Layanan. Fitur pra-GA tersedia "sebagaimana adanya" dan mungkin memiliki dukungan terbatas. Untuk mengetahui informasi selengkapnya, lihat deskripsi tahap peluncuran.

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

Jika ingin mengambil data aliran 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 kumpulan data output bergantung pada nilai parameter userOption. Jika userOption adalah NONE, output 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 level teratas sebagai kolom tabel. Misalnya, dokumen dalam koleksi MongoDB berisi kolom berikut:

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

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

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

Jika userOption adalah JSON, pipeline akan menyimpan dokumen dalam format JSON BigQuery. BigQuery memiliki dukungan bawaan untuk data JSON menggunakan jenis data JSON. Untuk mengetahui informasi selengkapnya, lihat Bekerja dengan data JSON di GoogleSQL.

Parameter template

Parameter yang diperlukan

• mongoDbUri : URI koneksi MongoDB dalam format mongodb+srv://:@..
• database : Database di MongoDB tempat koleksi dibaca. (Contoh: my-db).
• collection : Nama koleksi di dalam database MongoDB. (Contoh: my-collection).
• userOption : FLATTEN, JSON, atau NONE. FLATTEN meratakan dokumen ke tingkat tunggal. JSON menyimpan dokumen dalam format JSON BigQuery. NONE menyimpan seluruh dokumen sebagai STRING berformat JSON. Defaultnya adalah: NONE.
• outputTableSpec : Tabel BigQuery yang akan ditulis. Contoh, bigquery-project:dataset.output_table.

Parameter opsional

• KMSEncryptionKey : Kunci Enkripsi Cloud KMS untuk mendekripsi string koneksi uri mongodb. Jika kunci Cloud KMS diteruskan, string koneksi uri mongodb harus diteruskan dalam terenkripsi. (Contoh: projects/project-anda/locations/global/keyRings/keyring-anda/cryptoKeys/kunci-anda).
• filter : Filter Bson dalam format json. (Contoh: { "val": { $gt: 0, $lt: 9 }}).
• useStorageWriteApi : Jika true, pipeline akan menggunakan BigQuery Storage Write API (https://cloud.google.com/bigquery/docs/write-api). Nilai defaultnya adalah false. Untuk informasi selengkapnya, lihat Menggunakan Storage Write API (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
• useStorageWriteApiAtLeastOnce : Saat menggunakan Storage Write API, menentukan semantik tulis. Untuk menggunakan semantik minimal satu kali (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics), 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.
• bigQuerySchemaPath : Jalur Cloud Storage untuk skema JSON BigQuery. (Contoh: gs://your-bucket/your-schema.json).
• javascriptDocumentTransformGcsPath : URI Cloud Storage file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang akan digunakan. (Contoh: gs://bucket-anda/transformasi-anda/*.js).
• javascriptDocumentTransformFunctionName : Nama fungsi yang ditentukan pengguna (UDF) JavaScript yang akan digunakan. Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples). (Contoh: transformasi).

Fungsi yang ditentukan pengguna (UDF)

Secara opsional, Anda dapat memperluas template ini dengan menulis fungsi yang ditentukan pengguna (UDF) dalam 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 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

1. Buka halaman Create job from template Dataflow.
Buka Buat tugas dari template
2. Di kolom Nama tugas, masukkan nama tugas yang unik.
3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region defaultnya adalah us-central1.

   Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

4. Dari menu drop-down Dataflow template, pilih the MongoDB to BigQuery template.
5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
6. Klik Run job.

gcloud

Catatan: Untuk menggunakan Google Cloud CLI guna menjalankan template flex, Anda harus memiliki Google Cloud CLI versi 284.0.0, , atau yang lebih baru.

Di shell atau terminal, 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 tugas 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 template versi terbaru, yang tersedia di folder induk tanpa tanggal 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 dalam folder induk bertanggal masing-masing di bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru mungkin diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru untuk mencegah perubahan yang menyebabkan error ini 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, JSON, atau NONE.

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk mengetahui 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 tugas 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 template versi terbaru, yang tersedia di folder induk tanpa tanggal 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 dalam folder induk bertanggal masing-masing di bucket—gs://dataflow-templates-REGION_NAME/
  Perhatian: Template versi terbaru mungkin diupdate dengan perubahan yang dapat menyebabkan gangguan. Lingkungan produksi Anda harus menggunakan template yang disimpan di folder induk bertanggal terbaru untuk mencegah perubahan yang menyebabkan error ini 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, JSON, 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.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.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.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 = "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.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());
    }

    MongoDbIO.Read readDocuments =
        MongoDbIO.read()
            .withUri(mongoDbUri)
            .withDatabase(options.getDatabase())
            .withCollection(options.getCollection());

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

    pipeline
        .apply("Read Documents", readDocuments)
        .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.