MongoDB to BigQuery template

This template creates a batch pipeline that reads documents from MongoDB and writes them to BigQuery.

If you want to capture MongoDB change stream data, you can use the MongoDB to BigQuery (CDC) template.

Pipeline requirements

  • The target BigQuery dataset must exist.
  • The source MongoDB instance must be accessible from the Dataflow worker machines.

Output format

The format of the output records depends on the value of the userOption parameter. If userOption is NONE, the output has the following schema. The source_data field contains the document in JSON format.

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

If userOption is FLATTEN, the pipeline flattens the documents and writes the top-level fields as table columns. For example, suppose the documents in the MongoDB collection contain the following fields:

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

Using FLATTEN, the output has the following schema. The timestamp field is added by the template.

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

Template parameters

Required parameters

  • mongoDbUri : The MongoDB connection URI in the format mongodb+srv://:@..
  • database : Database in MongoDB to read the collection from. (Example: my-db).
  • collection : Name of the collection inside MongoDB database. (Example: my-collection).
  • userOption : FLATTEN, JSON, or NONE. FLATTEN flattens the documents to the single level. JSON stores document in BigQuery JSON format. NONE stores the whole document as a JSON-formatted STRING. Defaults to: NONE.
  • outputTableSpec : The BigQuery table to write to. For example, bigquery-project:dataset.output_table.

Optional parameters

  • KMSEncryptionKey : Cloud KMS Encryption Key to decrypt the mongodb uri connection string. If Cloud KMS key is passed in, the mongodb uri connection string must all be passed in encrypted. (Example: projects/your-project/locations/global/keyRings/your-keyring/cryptoKeys/your-key).
  • filter : Bson filter in json format. (Example: { "val": { $gt: 0, $lt: 9 }}).
  • useStorageWriteApi : If true, the pipeline uses the BigQuery Storage Write API (https://cloud.google.com/bigquery/docs/write-api). The default value is false. For more information, see Using the Storage Write API (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
  • useStorageWriteApiAtLeastOnce : When using the Storage Write API, specifies the write semantics. To use at-least-once semantics (https://beam.apache.org/documentation/io/built-in/google-bigquery/#at-least-once-semantics), set this parameter to true. To use exactly-once semantics, set the parameter to false. This parameter applies only when useStorageWriteApi is true. The default value is false.
  • bigQuerySchemaPath : The Cloud Storage path for the BigQuery JSON schema. (Example: gs://your-bucket/your-schema.json).
  • javascriptDocumentTransformGcsPath : The Cloud Storage URI of the .js file that defines the JavaScript user-defined function (UDF) to use. (Example: gs://your-bucket/your-transforms/*.js).
  • javascriptDocumentTransformFunctionName : The name of the JavaScript user-defined function (UDF) to use. For example, if your JavaScript function code is myTransform(inJson) { /*...do stuff...*/ }, then the function name is myTransform. For sample JavaScript UDFs, see UDF Examples (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples). (Example: transform).

User-defined function

Optionally, you can extend this template by writing a user-defined function (UDF) in JavaScript. The template calls the UDF for each input element. Element payloads are serialized as JSON strings.

To use a UDF, upload the JavaScript file to Cloud Storage and set the following template parameters:

ParameterDescription
javascriptDocumentTransformGcsPath The Cloud Storage location of the JavaScript file.
javascriptDocumentTransformFunctionName The name of the JavaScript function.

For more information, see Create user-defined functions for Dataflow templates.

Function specification

The UDF has the following specification:

  • Input: a MongoDB document.
  • Output: an object serialized as a JSON string. If userOption is NONE, the JSON object must include a property named _id that contains the document ID.

    • Run the template

    Console

    1. Go to the Dataflow Create job from template page.
      2. Go to Create job from template
    2. In the Job name field, enter a unique job name.
    3. Optional: For Regional endpoint, select a value from the drop-down menu. The default region is us-central1.

      For a list of regions where you can run a Dataflow job, see Dataflow locations.

    4. From the Dataflow template drop-down menu, select the MongoDB to BigQuery template.
    5. In the provided parameter fields, enter your parameter values.
    6. Click Run job.

    gcloud

    In your shell or terminal, run the 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

    Replace the following:

    • PROJECT_ID: the Google Cloud project ID where you want to run the Dataflow job
    • JOB_NAME: a unique job name of your choice
    • REGION_NAME: the region where you want to deploy your Dataflow job—for example, us-central1
    • VERSION: the version of the template that you want to use

      You can use the following values:

    • OUTPUT_TABLE_SPEC: your target BigQuery table name.
    • MONGO_DB_URI: your MongoDB URI.
    • DATABASE: your MongoDB database.
    • COLLECTION: your MongoDB collection.
    • USER_OPTION: FLATTEN or NONE.

    API

    To run the template using the REST API, send an HTTP POST request. For more information on the API and its authorization scopes, see 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",
   }
}

    Replace the following:

    • PROJECT_ID: the Google Cloud project ID where you want to run the Dataflow job
    • JOB_NAME: a unique job name of your choice
    • LOCATION: the region where you want to deploy your Dataflow job—for example, us-central1
    • VERSION: the version of the template that you want to use

      You can use the following values:

    • OUTPUT_TABLE_SPEC: your target BigQuery table name.
    • MONGO_DB_URI: your MongoDB URI.
    • DATABASE: your MongoDB database.
    • COLLECTION: your MongoDB collection.
    • USER_OPTION: FLATTEN or NONE.

    What's next