Venkatesh Shanbhag

Stanimira Vlaeva

On 1st October 2022, MongoDB and Google announced a set of open source [Dataflow templates](https://cloud.google.com/blog/products/data-analytics/mongodb-atlas-and-bigquery-dataflow-templates) for moving data between MongoDB and BigQuery to run analyses on BigQuery using BQML and to bring back inferences to MongoDB. Three templates were introduced as part of this release, including the MongoDB to BigQuery CDC (change data capture) template. 

![MongoDB change stream](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/change_stream_27de6ed2ac.png)

This template requires users to run the change stream on MongoDB, which will monitor inserts and updates on the collection. These changes will be captured and pushed to a [Pub/Sub topic](https://cloud.google.com/pubsub/docs/overview). The CDC template will create a job to read the data from the topic and get the changes, apply the transformation, and write the changes to BigQuery. The transformations will vary based on the user input while running the Dataflow job.

Alternatively, you can use a native Pub/Sub capability to set up a data pipeline between your MongoDB cluster and BigQuery. The Pub/Sub BigQuery subscription writes messages to an existing BigQuery table as they are received. Without the BigQuery subscription type, you need a pull or push subscription and a subscriber (such as Dataflow) that reads messages and writes them to BigQuery. 

This article explains how to set up the BigQuery subscription to process data read from a MongoDB change stream. As a prerequisite, you’ll need a MongoDB Atlas cluster.

> To set up a free tier cluster, you can register to MongoDB either from [Google Cloud Marketplace](https://console.cloud.google.com/marketplace/product/mongodb/mdb-atlas-self-service) or from the [registration page](https://www.mongodb.com/cloud/atlas/register). Follow the steps in the MongoDB documentation to configure the database user and network settings for your cluster. 

On Google Cloud, we will create a Pub/Sub topic, a BigQuery dataset, and a table before creating the BigQuery subscription.

## Create a BigQuery dataset


We’ll start by creating a new dataset for BigQuery in [Google Cloud console](https://console.cloud.google.com/bigquery).

Then, add a new table in your dataset. Define it with a name of your choice and the following schema:


| Field name | Type |
| --- | --- |
| id | STRING |
| source_data | STRING |
| Timestamp | STRING |

![Create a new dataset and table in BigQuery](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/source_and_destination_fb83305dc6.png)

## Configure Google Cloud Pub/Sub

Next, we’ll configure a Pub/Sub schema and topic to ingest the messages from our MongoDB change stream. Then, we’ll create a subscription to write the received messages to the BigQuery table we just created.

For this section, we’ll use the Google Cloud Pub/Sub API. Before proceeding, make sure you have enabled the API for your project.

### Define a Pub/Sub schema 

From the Cloud Pub/Sub UI, Navigate to _[Create Schema](https://console.cloud.google.com/cloudpubsub/schema/create)_.

Provide an appropriate identifier, such as “mdb-to-bq-schema,” to your schema. Then, select “Avro” for the type. Finally, add the following definition to match the fields from your BigQuery table:

```json
{
 "type" : "record",
 "name" : "Avro",
 "fields" : [
   {
     "name" : "id",
     "type" : "string"
   },
   {
     "name" : "source_data",
     "type" : "string"
   },
   {
     "name" : "Timestamp",
     "type" : "string"
   }
 ]
}
```

![Create a Cloud Pub/Sub schema](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_schema_7db8b97369.png)

### Create a Pub/Sub topic

From the sidebar, navigate to “Topics” and click on Create a topic. 

Give your topic an identifier, such as “MongoDBCDC.” Enable the Use a schema field and select the schema that you just created. Leave the rest of the parameters to default and click on _Create Topic_.

![Create a Cloud Pub/Sub topic](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_topic_4948d0ebc9.png)

### Subscribe to topic and write to BigQuery

From inside the topic, click on _Create new subscription_. Configure your subscription in the following way:

- Provide a subscription ID — for example, “mdb-cdc.”
- Define the Delivery type to _Write to BigQuery_.
- Select your BigQuery dataset from the dropdown.
- Provide the name of the table you created in the BigQuery dataset.
- Enable _Use topic schema_.

You need to have a `bigquery.dataEditor` role on your service account to create a Pub/Sub BigQuery subscription. To grant access using the [`bq` command line tool](https://cloud.google.com/bigquery/docs/bq-command-line-tool), run the following command:

```sh
bq add-iam-policy-binding \
    --member="serviceAccount:service<project-number>@gcp-sa-pubsub.iam.gserviceaccount.com" \
    --role=roles/bigquery.dataEditor \
    -t "<dataset>.<table>"
```


Keep the other fields as default and click on _Create subscription_.

![Create a Cloud Pub/Sub subscription](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_subscription_f693420e69.png)

## Set up a change stream on a MongoDB cluster

Finally, we’ll set up a change stream that listens for new documents inserted in our MongoDB cluster. 

We’ll use Node.js but you can adapt the code to a programming language of your choice. Check out the [Google Cloud documentation](https://cloud.google.com/pubsub/docs/publisher#using-schema) for more Pub/Sub examples using a variety of languages. You can find the source code of this example in the dedicated [GitHub repository](https://github.com/mongodb-developer/MongoDB-Change-Stream-to-Google-Cloud-PubSub).

First, set up a new Node.js project and install the following dependencies.

```sh
npm install mongodb @google-cloud/pubsub avro-js
```

Then, add an Avro schema, matching the one we created in Google Cloud Pub/Sub:

**./document-message.avsc**
```json
{
 "type": "record",
 "name": "DocumentMessage",
 "fields": [
   {
     "name": "id",
     "type": "string"
   },
   {
     "name": "source_data",
     "type": "string"
   },
   {
     "name": "Timestamp",
     "type": "string"
   }
 ]
}
```

Then create a new JavaScript module — `index.mjs`. Start by importing the required libraries and setting up your MongoDB connection string and your Pub/Sub topic name. If you don’t already have a MongoDB cluster, you can create one for free in [MongoDB Atlas](https://www.mongodb.com/cloud/atlas/register).

**./index.mjs**
```js
import { MongoClient } from 'mongodb';
import { PubSub } from '@google-cloud/pubsub';
import avro from 'avro-js';
import fs from 'fs';
 
const MONGODB_URI = '<mongodb-connection-string>';
const PUB_SUB_TOPIC = 'projects/<project-name>/topics/<topic-name>';
```

After this, we can connect to our MongoDB instance and set up a [change stream event listener](https://www.mongodb.com/docs/manual/reference/method/db.collection.watch/). Using an aggregation pipeline, we’ll watch only for “insert” events on the specified collection. We’ll also define a 60-second timeout before closing the change stream.

**./index.mjs**
```js
let mongodbClient;
try {
   mongodbClient = new MongoClient(MONGODB_URI);
   await monitorCollectionForInserts(mongodbClient, 'my-database', 'my-collection');
} finally {
   mongodbClient.close();
}
 
async function monitorCollectionForInserts(client, databaseName, collectionName, timeInMs) {
   const collection = client.db(databaseName).collection(collectionName);
   // An aggregation pipeline that matches on new documents in the collection.
   const pipeline = [ { $match: { operationType: 'insert' } } ];
   const changeStream = collection.watch(pipeline);
 
   changeStream.on('change', event => {
       const document = event.fullDocument;
       publishDocumentAsMessage(document, PUB_SUB_TOPIC);
   });
 
   await closeChangeStream(timeInMs, changeStream);
}
 
function closeChangeStream(timeInMs = 60000, changeStream) {
   return new Promise((resolve) => {
       setTimeout(() => {
           console.log('Closing the change stream');
           changeStream.close();
           resolve();
       }, timeInMs)
   })
};
```


Finally, we’ll define the `publishDocumentAsMessage()` function that will:

1. Transform every MongoDB document received through the change stream.
1. Convert it to the data buffer following the Avro schema.
1. Publish it to the Pub/Sub topic in Google Cloud.

```js
async function publishDocumentAsMessage(document, topicName) {
   const pubSubClient = new PubSub();
   const topic = pubSubClient.topic(topicName);
 
   const definition = fs.readFileSync('./document-message.avsc').toString();
   const type = avro.parse(definition);
 
   const message = {
       id: document?._id?.toString(),
       source_data: JSON.stringify(document),
       Timestamp: new Date().toISOString(),
   };
 
   const dataBuffer = Buffer.from(type.toString(message));
   try {
       const messageId = await topic.publishMessage({ data: dataBuffer });
       console.log(`Avro record ${messageId} published.`);
   } catch(error) {
       console.error(error);
   }
}
```

Run the file to start the change stream listener:

```sh
node ./index.mjs
```

Insert a new document in your MongoDB collection to watch it go through the data pipeline and appear in your BigQuery table!

## Summary

There are multiple ways to load the change stream data from MongoDB to BigQuery and we have shown how to use the BigQuery subscription on Pub/Sub. The change streams from MongoDB are monitored, captured, and later written to a Pub/Sub topic using Java libraries.

The data is then written to BigQuery using BigQuery subscription. The datatype for the BigQuery table is set using Pub/Sub schema. Thus, the change stream data can be captured and written to BigQuery using the BigQuery subscription capability of Pub/Sub.

## Further reading

1. A data pipeline for [MongoDB Atlas and BigQuery](https://cloud.google.com/blog/products/data-analytics/mongodb-atlas-and-bigquery-dataflow-templates) using Dataflow.
1. [Setup your first MongoDB cluster](https://www.mongodb.com/docs/atlas/billing/gcp-self-serve-marketplace/) using Google Marketplace.
1. Run analytics using BigQuery using [BigQuery ML](https://cloud.google.com/bigquery-ml/docs).
1. [How to publish](https://cloud.google.com/pubsub/docs/publisher#using-schema) a message to a topic with schema.

Learn how to set up a data pipeline from your MongoDB database to BigQuery using change streams and Google Cloud Pub/Sub.

Create a Data Pipeline for MongoDB Change Stream Using Pub/Sub BigQuery Subscription

Ozan Güzeldereli

Many enterprise customers of MongoDB and Google Cloud have the core operation workload running on MongoDB and run their analytics on BigQuery. To make it seamless to move the data between MongoDB and BigQuery, MongoDB introduced Google [Dataflow templates](https://cloud.google.com/blog/products/data-analytics/mongodb-atlas-and-bigquery-dataflow-templates). Though these templates cater to most of the common use cases, there is still some effort required to set up the change stream (CDC) Dataflow template. Setting up the CDC requires users to create their own custom code to monitor the changes happening on their MongoDB Atlas collection. Developing custom codes is time-consuming and requires a lot of time for development, support, management, and operations.

![MongoDB to BigQuery pipeline using Dataflow templates](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_bigquery_pipeline_48206d2122.png)

Overcoming the additional effort required to set up CDCs for MongoDB to BigQuery Dataflow templates can be achieved using Confluent Cloud. Confluent is a full-scale data platform capable of continuous, real-time processing, integration, and data streaming across any infrastructure. Confluent provides pluggable, declarative data integration through its connectors. With Confluent’s MongoDB source connectors, the process of creating and deploying a module for CDCs can be eliminated. Confluent Cloud provides a MongoDB Atlas source connector that can be easily configured from Confluent Cloud, which will read the changes from the MongoDB source and publish those changes to a topic. Reading from MongoDB as source is the part of the solution that is further enhanced with a Confluent BigQuery sink connector to read changes that are published to the topic and then writing to the BigQuery table.  

![Architecture for MongoDB to BigQuery pipeline using Confluent connectors](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/image1_d60f0a96c7.jpg)

This article explains how to set up the MongoDB cluster, Confluent cluster, and Confluent MongoDB Atlas source connector for reading changes from your MongoDB cluster, BigQuery dataset, and Confluent BigQuery sink connector.

As a prerequisite, we need a MongoDB Atlas cluster, Confluent Cloud cluster, and Google Cloud account. If you don’t have the accounts, the next sections will help you understand how to set them up.

### Set up your MongoDB Atlas cluster
To set up your first MongoDB Atlas cluster, you can register for MongoDB either from Google [Marketplace](https://console.cloud.google.com/marketplace/product/mongodb/mdb-atlas-self-service) or from the [registration](https://www.mongodb.com/cloud/atlas/register) page. Once registered for MongoDB Atlas, you can [set up](https://www.mongodb.com/docs/atlas/tutorial/create-new-cluster/#select-a-database-deployment-type) your first free tier Shared M0 cluster. Follow the steps in the MongoDB documentation to configure the [database user](https://www.mongodb.com/docs/atlas/security-add-mongodb-users/) and [network](https://www.mongodb.com/docs/atlas/security/ip-access-list/) settings for your cluster. 

![MongoDB Atlas Sandbox cluster set up](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_atlas_sandbox_cluster_29730ba626.png)

Once the cluster and access setup is complete, we can load some sample data to the cluster. Navigate to “browse collection” from the Atlas homepage and click on “Create Database.” Name your database “Sample_company” and collection “Sample_employee.”

Insert your first document into the database:

```
{
"Name":"Jane Doe",
"Address":{
"Phone":{"$numberLong":"999999"},
"City":"Wonderland"
}
}
}
```

## Set up a BigQuery dataset on Google Cloud
As a prerequisite for setting up the pipeline, we need to create a dataset in the same region as that of the Confluent cluster. Please go through the Google [documentation](https://cloud.google.com/bigquery/docs/datasets#console) to understand how to create a dataset for your project. Name your dataset “Sample_Dataset.”

## Set up the Confluent Cloud cluster and connectors
After setting up the MongoDB and BigQuery datasets, Confluent will be the platform to build the data pipeline between these platforms. 

To sign up using Confluent Cloud, you can either go to the [Confluent website](https://www.confluent.io/get-started/) or register from [Google Marketplace](https://console.cloud.google.com/marketplace/product/confluent-prod/apache-kafka-on-confluent-cloud). New signups receive $400 to spend during their first 30 days and a credit card is not required. To create the cluster, you can follow the first step in the [documentation](https://docs.confluent.io/cloud/current/get-started/index.html#step-1-create-a-ak-cluster-in-ccloud). **One important thing to consider is that the region of the cluster should be the same region of the GCP BigQuery cluster.**

### Set up your MongoDB Atlas source connector on Confluent
Depending on the settings, it may take a few minutes to provision your cluster, but once the cluster has provisioned, we can get the sample data from MongoDB cluster to the Confluent cluster.

Confluent’s MongoDB Atlas Source [connector](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-source.html) helps to read the change stream data from the MongoDB database and write it to the topic. This connector is fully managed by Confluent and you don’t need to operate it. To set up a connector, navigate to [Confluent Cloud](https://confluent.cloud/) and search for the MongoDB Atlas source connector under “Connectors.” The [connector documentation](https://docs.confluent.io/cloud/current/connectors/cc-mongo-db-source.html#quick-start) provides the steps to provision the connector. 

Below is the sample configuration for the MongoDB source connector setup.

1. For **Topic selection**, leave the prefix empty.
2. Generate **Kafka credentials** and click on “Continue.”
3. Under Authentication, provide the details:
    1. Connection host: Only provide the MongoDB Hostname in format “mongodbcluster.mongodb.net.”
    2. Connection user: MongoDB connection user name.
    3. Connection password: Password of the user being authenticated.
    4. Database name: **sample_database** and collection name: **sample_collection**.
4. Under configuration, select the output Kafka record format as **JSON_SR** and click on “Continue.”
5. Leave sizing to default and click on “Continue.”
6. Review and click on “Continue.”

![Confluent connector configuration for MongoDB source connector](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/confluent_connector_configuration_43590e9db1.png)

```
{
  "name": "MongoDbAtlasSourceConnector",
  "config": {
    "connector.class": "MongoDbAtlasSource",
    "name": "MongoDbAtlasSourceConnector",
    "kafka.auth.mode": "KAFKA_API_KEY",
    "kafka.api.key": "****************",
    "kafka.api.secret": "****************************************************************",
    "connection.host": "mongodbcluster.mongodb.net",
    "connection.user": "testuser",
    "connection.password": "*********",
    "database": "Sample_Company",
    "collection": "Sample_Employee",
    "output.data.format": "JSON_SR",
    "publish.full.document.only": "true",
    "tasks.max": "1"
  }
}
```

### Set up Confluent Cloud: BigQuery sink connector
After setting up our BigQuery, we need to provision a sink connector to sink the data from Confluent Cluster to Google BigQuery. The Confluent Cloud to BigQuery Sink connector can stream table records from Kafka topics to Google BigQuery. The table records are streamed at high throughput rates to facilitate analytical queries in real time.

To set up the Bigquery sink connector, [follow the steps](https://docs.confluent.io/cloud/current/connectors/cc-gcp-bigquery-sink.html#using-the-ccloud-console) in their documentation. 

```
{
  "name": "BigQuerySinkConnector_0",
  "config": {
    "topics": "AppEngineTest.emp",
    "input.data.format": "JSON_SR",
    "connector.class": "BigQuerySink",
    "name": "BigQuerySinkConnector_0",
    "kafka.auth.mode": "KAFKA_API_KEY",
    "kafka.api.key": "****************",
    "kafka.api.secret": "****************************************************************",
    "keyfile": "******************************************************************************
—--
***************************************",
    "project": "googleproject-id",
    "datasets": "Sample_Dataset",
    "auto.create.tables": "true",
    "auto.update.schemas": "true",
    "tasks.max": "1"
  }
}
```

To see the data being loaded to BigQuery, make some changes on the MongoDB collection. Any inserts and updates will be recorded from MongoDB and pushed to BigQuery. 

Insert below document to your MongoDB collection using MongoDB Atlas UI. (Navigate to your collection and click on “INSERT DOCUMENT.”)

```
{
"Name":"John Doe",
"Address":{
"Phone":{"$numberLong":"8888888"},
"City":"Narnia"
}
}
}
```

## Summary
MongoDB and Confluent are positioned at the heart of many modern data architectures that help developers easily build robust and reactive data pipelines that stream events between applications and services in real time. In this example, we provided a template to build a pipeline from MongoDB to Bigquery on Confluent Cloud. Confluent Cloud provides more than 200 connectors to build such pipelines between many solutions. Although the solutions change, the general approach is using those connectors to build pipelines.

### What's next?

1. To understand the features of Confluent Cloud managed MongoDB sink and source connectors, you can watch [this webinar](https://www.confluent.io/resources/online-talk/build-fully-managed-data-pipelines-with-mongodb-atlas-and-confluent/). 
2. Learn more about the Bigquery sink [connector](https://www.confluent.io/blog/streaming-etl-sfdc-data-for-real-time-customer-analytics/?_ga=2.116016451.45433106.1669022889-1031176121.1668508664).
3. A data pipeline for [MongoDB Atlas and BigQuery](https://cloud.google.com/blog/products/data-analytics/mongodb-atlas-and-bigquery-dataflow-templates) using Dataflow.
4. [Set up your first MongoDB cluster](https://www.mongodb.com/docs/atlas/billing/gcp-self-serve-marketplace/) using Google Marketplace.
5. Run analytics using BigQuery using [BigQuery ML](https://cloud.google.com/bigquery-ml/docs).



Learn how to set up a data pipeline from your MongoDB database to BigQuery using the Confluent connector.

Streaming Data from MongoDB to BigQuery Using Confluent Connectors

Dianna Annie Jenosh

Many enterprise customers using MongoDB Atlas as their core operational database also use BigQuery for their Batch and AI/ML based analytics, making it pivotal for seamless transfer of data between these entities. Since the announcement of the [Dataflow templates](https://cloud.google.com/blog/products/data-analytics/mongodb-atlas-and-bigquery-dataflow-templates) (in October Of 2022) on moving data between MongoDB and BigQuery, we have seen a lot of interest from customers as it made it effortless for an append-only, one-to-one migration of data. Though the three Dataflow templates provided cater to most of the common use cases, there was also a demand to be able to do transformations as part of these templates.

We are excited to announce the addition of the ability to write your own user-defined functions (UDFs) in these Dataflow pipelines! This new feature allows you to use UDFs in JavaScript to transform and analyze data within BigQuery. With UDFs, you can define custom logic and business rules that can be applied to your data as it is being processed by Dataflow. This allows you to perform complex transformations like transforming fields, concatenating fields, deleting fields, converting embedded documents to separate documents, etc. These UDFs take unprocessed documents as input parameters and return the processed documents as output.

To use UDFs with BigQuery Dataflow, simply write your JavaScript function and store it in the Google cloud storage bucket. Use the Dataflow templates’ optional parameter to read these UDFs while running the templates. The function will be executed on the data as it is being processed, allowing you to apply custom logic and transformations to your data during the transfer.

## How to set it up
Let’s have a quick look at how to set up a sample UDF to process (transform a field, flatten an embedded document, and delete a field) from an input document before writing the processed data to BigQuery.

### Set up MongoDB 

1. MongoDB Atlas setup through [registration](https://www.mongodb.com/cloud/atlas/register).
2. MongoDB Atlas setup through [GCP Marketplace](https://console.cloud.google.com/marketplace/product/mongodb/mdb-atlas-self-service?utm_source=mongodb_devrel&utm_medium=static_site&utm_campaign=devrel&utm_id=mongodb). (MongoDB Atlas is available pay as you go in the GC marketplace).
3. Create your MongoDB [cluster](https://www.mongodb.com/docs/atlas/tutorial/create-new-cluster/).
4. Click on **Browse collections** and click on **+Create Database**.

![MongoDB Atlas cluster landing page](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/mongodb_atlas_cluster_landing_page_ad369de42d.png)

5: Name your database **Sample_Company** and collection **Sample_Employee**.
![MongoDB Atlas create collection step](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_database_30d4e827a2.png)
6: Click on **INSERT DOCUMENT**.
![Insert document to MongoDB collection from Atlas UI](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/insert_document_2e7d4f16ef.png)
Copy and paste the below document and click on **Insert**.
```
{
  "Name":"Venkatesh",
  "Address":{"Phone":{"$numberLong":"123455"},"City":"Honnavar"},
  "Department":"Solutions Consulting",
  "Direct_reporting": "PS"
}
```
7: To have authenticated access on the MongoDB Sandbox cluster from Google console, we need to create database users.

Click on the **Database Access** from the left pane on the Atlas Dashboard. 

Choose to **Add New User** using the green button on the left. Enter the username `appUser` and password `appUser123`. We will use built-in roles; click **Add Default Privileges** and in the **Default Privileges** section, add the roles readWriteAnyDatabase. Then press the green **Add User** button to create the user.

![Create a database user on MongoDB Atlas](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_database_user_774995251a.png)

8: Whitelist the IPs.

For the purpose of this demo, we will allow access from any ip, i.e 0.0.0.0/0. However, this is not recommended for a production setup, where the recommendation will be to use VPC Peering and private IPs.

![IP whitelisting on MongoDB Atlas to allow external access](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/ip_whitelisting_f5bc31c2bb.png)

![Whitelist 0.0.0.0/0](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/whitelisting_c0f1df1e03.png)

### Set up Google Cloud

1. Create a cloud storage bucket.
2. On your local machine, create a Javascript file **transform.js** and add below sample code.

```
function transform(inputDoc) {
   var outputDoc = new Object();
   inputDoc["City"] = inputDoc["Address"]["City"];
   delete doc.Address;
   outputDoc = doc;
   return returnObj;
}
```

This function will read the document read from MongoDB using the Apache beam MongoDB IO connector. Flatten the embedded document Address/City to City. Delete the Address field and return the updated document.

3: [Upload](https://cloud.google.com/storage/docs/uploading-objects) the javascript file to the Google Cloud storage bucket.

4: [Create](https://cloud.google.com/bigquery/docs/datasets#create-dataset) a BigQuery Dataset in your project in the region close to your physical location.

5: Create a [Dataflow pipeline](https://console.cloud.google.com/dataflow).

a. Click on the **Create Job from the template** button at the top.

b. Job Name: **mongodb-udf**.

c. Region: Same as your BigQuery dataset region.

d. MongoDB connection URI: Copy the connection URI for connecting applications from MongoDB Atlas.

e. MongoDB Database: **Sample_Company**.

f. MongoDB Collection: **Sample_Employee**.

g. BigQuery Destination Table: Copy the destination table link from the BigQuery

h. Dataset details page in format: bigquery-project:**sample_dataset.sample_company**.

i. User Option: **FLATTEN**.

j. Click on **show optional parameters**.

k. Cloud storage location of your Javascript UDF: Browse your UDF file loaded to bucket location. This is the new feature that allows running the UDF and applies the transformations before inserting into BigQuery.

l. Name of your Javascript function: **transform**.

![Create Dataflow Job with UDF on Google Cloud console](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/create_dataflow_job_cfdbd2357e.png)

6: Click on **RUN JOB** to start running the pipeline. Once the pipeline finishes running,  your graph should show **Succeeded** on each stage as shown below.

![Dataflow Job Graph view on Google Cloud console](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/dataflow_job_graph_ff48458c96.png)

7: After completion of the job, you will be able to see the transformed document inserted into BigQuery.

![Document read from MongoDB flattened and loaded to BigQuery](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/document_read_ed7e24b805.png)

## Conclusion
In this blog, we introduced UDFs to MongoDB to BigQuery Dataflow templates and their capabilities to transform the documents read from MongoDB using custom user defined Javascript functions stored on Google Cloud storage buckets. This blog also includes a simple tutorial on how to set up MongoDB Atlas, Google Cloud, and the UDFs.

### Further reading

* A data pipeline for [MongoDB Atlas and BigQuery using Dataflow](https://cloud.google.com/blog/products/data-analytics/mongodb-atlas-and-bigquery-dataflow-templates).
* A data pipeline for [MongoDB Atlas and BigQuery using the Confluent](https://www.mongodb.com/developer/products/atlas/mongodb-bigquery-pipeline-using-confluent/) connector.
* Run analytics using BigQuery using [BigQuery ML](https://cloud.google.com/bigquery-ml/docs).
* [Set up your first MongoDB cluster](https://www.mongodb.com/docs/atlas/billing/gcp-self-serve-marketplace/) using Google Marketplace.


Learn how to transform the MongoDB Documents using user-defined JavaScript functions in Dataflow templates.

UDF Announcement for MongoDB to BigQuery Dataflow Templates

In an era of widespread digitalization, businesses operating in critical sectors such as healthcare, banking and finance, and government face an ever-increasing threat of data breaches and cyber-attacks. Ensuring the security of data is no longer just a matter of compliance but has become a top priority for businesses to safeguard their reputation, customer trust, and financial stability. However, maintaining the privacy and security of sensitive data while still enabling seamless access to services within a virtual private cloud (VPC) is a complex challenge that requires a robust solution. That’s where MongoDB’s Private Service Connect (PSC) on Google Cloud comes in. As a cloud networking solution, it provides secure access to services within a VPC using private IP addresses. PSC is also a powerful tool to protect businesses from the ever-evolving threat landscape of data security. 

## What is PSC (Private Service Connect)?

PSC simplifies how services are being securely and privately consumed. It allows easy implementation of private endpoints for the service consumers to connect privately to service producers across organizations and eliminates the need for virtual private cloud peering. The effort needed to set up private connectivity between MongoDB and Google consumer project is reduced with the PSC.

MongoDB announced the [support](https://www.mongodb.com/blog/post/announcing-google-private-service-connect-psc-integration-mongodb-atlas) for Google Cloud [Private Service Connect](https://cloud.google.com/vpc/docs/private-service-connect) (PSC) in November 2021. PSC was added as a new option to access MongoDB securely from Google Cloud without exposing the customer traffic to the public internet. With PSC, customers will be able to achieve one-way communication with MongoDB. In this article, we are going to introduce the new features of PSC and MongoDB integration.

## PSC Interconnect support

Connecting MongoDB from the on-prem machines is made easy using PSC Interconnect support. PSC Interconnect allows traffic from on-prem devices to reach PSC endpoints in the same region as the Interconnect. This is also a transparent update with no API changes.

There are no additional actions required by the customer to start using their Interconnect with PSC. Once Interconnect support has been rolled out to the customer project, then traffic from the Interconnect will be able to reach PSC endpoints and in turn access the data from MongoDB using service attachments.

## Google Cloud multi-region support

Private Service Connect now provides multi-region support for MongoDB Atlas clusters, enabling customers to connect to MongoDB instances in different regions securely. With this feature, customers can ensure high availability even in case of a regional failover. To achieve this, customers need to set up the service attachments in all the regions that the cluster will have its nodes on. Each of these service attachments are in turn connected to Google Cloud service endpoints.

![A connection made from us-east1 will be able to connect to region us-west1 in case of the primary failover from region us-east1 to us-west1.](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/connection_9d618950d6.png)

## MongoDB multi-cloud support

Customers who have their deployment on multiple regions spread across multiple clouds can now utilize MongoDB PSC to connect to the Google Cloud nodes in their deployment. The additional requirement is to set up the private link for the other nodes to make sure that the connection could be made to the other nodes from their respective cloud targets. 

## Wrap-up

In conclusion, Private Service Connect has come a long way from its initial release. Now, PSC on MongoDB supports connection from on-prem using Interconnect and also connects to multiple regions across MongoDB clusters spread across Google Cloud regions or multi-cloud clusters securely using Global access.

1. Learn how to set up PSC multi region for MongoDB Atlas with [codelabs tutorials](https://codelabs.developers.google.com/codelabs/psc-mongo-globalaccess#0).
2. You can subscribe to MongoDB Atlas using [Google Cloud Marketplace](https://console.cloud.google.com/marketplace/product/mongodb/mdb-atlas-self-service).
3. You can [sign up](https://mongodb.com/cloud/atlas/register?utm_campaign=devrel&utm_source=cross-post&utm_medium=cta&utm_content=psc_announcement&utm_term=stanimira.vlaeva) for MongoDB using the registration page.
4. Learn more about [Private Service Connect](https://cloud.google.com/vpc/docs/private-service-connect).
5. Read the [PSC announcement for MongoDB](https://mongodb.com/blog/post/announcing-google-private-service-connect-psc-integration-mongodb-atlas?utm_campaign=devrel&utm_source=cross-post&utm_medium=cta&utm_content=psc_integration&utm_term=stanimira.vlaeva).

PSC is a cloud networking solution that provides secure access to services within a VPC. Read about the newly announced support for PSC Interconnect and Global access for MongoDB.

Venkatesh Shanbhag

Content by Venkatesh Shanbhag

Create a Data Pipeline for MongoDB Change Stream Using Pub/Sub BigQuery Subscription

Streaming Data from MongoDB to BigQuery Using Confluent Connectors

UDF Announcement for MongoDB to BigQuery Dataflow Templates

Introducing PSC Interconnect and Global Access for MongoDB Atlas