Venkatesh Shanbhag

Partner Solution Architect

Partner

Zi Wang

Partner Engineer at Google Cloud

Maruti C

Global Partner Engineering Lead, Google

To empower enterprises that strive to transform their data into insights, BigQuery has emerged as a powerful, scalable, cloud-based data warehouse solution offered by Google Cloud Platform (GCP). Its cloud-based approach allows efficient data management and manipulation, making BigQuery a game-changer for businesses seeking advanced data insights. Notably, one of BigQuery’s standout features is its seamless integration with Spark-based data processing that enables users to further enhance their queries. Now, leveraging BigQuery APIs, users can create and execute Spark stored procedures, which are reusable code modules that can encapsulate complex business logic and data transformations. This feature helps data engineers, data scientists, and data analysts take advantage of BigQuery’s advanced capabilities and Spark’s robust data processing capabilities.

MongoDB, a developer data platform, is a popular choice for storing and managing operational data for its scalability, performance, flexible schema, and real-time capabilities (change streams and aggregation). By combining the capabilities of BigQuery with the versatility of Apache Spark and the flexibility of MongoDB, you can unlock a powerful data processing pipeline.

Apache Spark is a powerful open-source distributed computing framework that excels at processing large amounts of data quickly and efficiently. It supports a wide range of data formats, including structured, semi-structured, and unstructured data, making it an ideal choice for integrating data from various sources, such as MongoDB.

BigQuery Spark stored procedures are routines that are executed within the BigQuery environment. These procedures can perform various tasks, such as data manipulation, complex calculations, and even external data integration. They provide a way to modularize and reuse code, making it easier to maintain and optimize data processing workflows. Spark stored procedures use the [serverless Spark](https://cloud.google.com/solutions/spark) engine that enables serverless, autoscaling Spark. However, you don’t need to enable Dataproc APIs or be charged for Dataproc when you leverage this new capability. 

Let's explore how to extend BigQuery’s data processing to Apache Spark, and integrate MongoDB with BigQuery to effectively facilitate data movement between the two platforms.

## Connecting them together

![Data movement between MongoDB and BigQuery using Spark stored procedure running on serverless Spark][1]

This tutorial guides you through creating a PySpark procedure using the BigQuery editor.

Before we start with the Spark stored procedure setup, you need to [upload](https://cloud.google.com/storage/docs/uploading-objects#uploading-an-object) the [MongoDB Spark connector JAR](https://repo1.maven.org/maven2/org/mongodb/spark/) file to Google Cloud Storage to connect and read from MongoDB Atlas. Copy and save the gsutil URI for the JAR file that will be used in upcoming steps.

![Copy the path for gsutil URI][2]

1. As a prerequisite for completing the tutorial, you need to [set up](https://www.mongodb.com/docs/atlas/getting-started/) a MongoDB Atlas cluster with [sample data](https://www.mongodb.com/docs/guides/atlas/sample-data/) loaded to it. 
2. Navigate to the [BigQuery page](https://console.cloud.google.com/bigquery) on the Google Cloud console.
3. [Create](https://cloud.google.com/bigquery/docs/datasets#create-dataset) a **BigQuery dataset** with the name **spark_run**.
4. You will type the PySpark code directly into the query editor. To create a PySpark stored procedure, click on **Create Pyspark Procedure**, and then select **Create PySpark Procedure**.

![Creating PySPark procedure from BigQuery explorer][3]

5. To set options, click **More** > **PySpark Options**, and then do the following:

    1. Specify the location where you want to run the PySpark code.

    2. In the **Connection** field, click on **CREATE NEW CONNECTION** and enter the below values for each field.

        1. Connection Type > Apache Spark

        2. Connection id > Name it as **mongodb-to-bigquery** for the sake of this tutorial.
        3. Leave the other option empty and click on **Create**.

    3. In the Stored procedure invocation section select **Set a dataset for stored procedure invocation** to **spark_run.**
    4. Click on the advanced options. Copy the gsutil URI name copied at the beginning of the setup and paste in the **JAR files in the pre-requisite**. Leave the other options empty. Press enter and click on **Save**.

6. Open a new tab and go to BigQuery. Navigate to the **External connections** > Find the **mongodb-to-bigquery Connection** > Copy the **Service account id**. [Grant](https://cloud.google.com/iam/docs/grant-role-console#grant_an_iam_role) BigQuery Storage Admin, Secret Manager Secret Accessor, and Storage Object Admin access to this service account from IAM. 

![External connections created using Spark stored procedures][4]

7. (Optional) [Add your username and password](https://cloud.google.com/secret-manager/docs/creating-and-accessing-secrets#create) into Google Cloud Secret Manager, or you can hardcode it in the MongoDB URI string itself. 
8. Copy the below Python script in the PySpark procedure editor and click on **RUN**. The snippet takes around two to three minutes to complete. The below script will create a new table under dataset **spark_run** with the name **sample_mflix_comments**.

```python
from pyspark.sql import SparkSession
from google.cloud import secretmanager

def access_secret_version(secret_id, project_id):
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{project_id}/secrets/{secret_id}/versions/1"
    response = client.access_secret_version(request={"name": name})
    payload = response.payload.data.decode("UTF-8")
    return payload
# Update project_number, username_secret_id and password_secret_id, comment them out if you did not create the secrets earlier 

project_id = "<Your project number, 12 digit number>"
username_secret_id = "<Your username secret id>"
password_secret_id = "<Your password secret id>"

username = access_secret_version(username_secret_id, project_id)
password = access_secret_version(password_secret_id, project_id)

  # Update the mongodb_uri directly if with your username and password if you did not create a secret from Step 7, update the hostname with your hostname
mongodb_uri = "mongodb+srv://"+username+":"+password+"@<hostname>/sample_mflix.comments"

my_spark = SparkSession \
  .builder \
  .appName("myApp") \
  .config("spark.mongodb.read.connection.uri", mongodb_uri) \
  .config("spark.mongodb.write.connection.uri", mongodb_uri) \
  .getOrCreate()


dataFrame = my_spark.read.format("mongodb").option("database", "sample_mflix").option("collection", "comments").load()

dataFrame.show()

# Saving the data to BigQuery
dataFrame.write.format("bigquery") \
  .option("writeMethod", "direct") \
    .save("spark_run.sample_mflix_comments")
```

![Stored procedure on Google BigQuery Explorer][5]

9. Navigate to the spark_run dataset to validate the data is loaded from MongoDB Atlas to BigQuery under the table with the name **sample_mflix_comments**.
10. Now that the data is in BigQuery, leverage BQML to run some Generative AI on the new MongoDB data in BigQuery. 
    1. Create a connection with the name **gentext-conn,** using either the [console](https://cloud.google.com/bigquery/docs/create-cloud-resource-connection#create-cloud-resource-connection) or bq command line with connection type as CLOUD_RESOURCE.

```
!bq mk \
  --connection \
  --location=US \
  --project_id=<GCP Project id> \
  --connection_type=CLOUD_RESOURCE gentext-conn 
```

11. To grant IAM permissions to access Vertex AI from BigQuery, navigate to **External connections** > Find the **gettext-conn** connection > Copy the **Service account id**. Grant the **Vertex AI User** access to this service account from **IAM**.
12. Create a model using the CREATE MODEL command.

```
CREATE OR REPLACE MODEL `gcp-pov.spark_run.llm_model`
REMOTE WITH CONNECTION `us.gentext-conn`
OPTIONS (ENDPOINT = 'gemini-pro');
```

13. Run the SQL command against the BigQuery table. This query allows the user to extract the host name from the email leveraging the Gemini Pro model. The resulting [output](https://cloud.google.com/bigquery/docs/reference/standard-sql/bigqueryml-syntax-generate-text#output) includes the response and safety attributes.

```
SELECT prompt,ml_generate_text_result
FROM
ML.GENERATE_TEXT( MODEL `gcp-pov.spark_run.llm_model`,
  (
    SELECT CONCAT('Extract the host name from the email: ', email) AS prompt,
      *  FROM  `gcp-pov.spark_run.sample_mflix_comments`
    LIMIT 5),
  STRUCT(
    0.9 AS temperature,
    100 AS max_output_tokens
    )
  );
```

14. Here is the [sample output](https://cloud.google.com/bigquery/docs/reference/standard-sql/bigqueryml-syntax-generate-text#output) showing the prompt as well as the response. The prompt parameter provides the text for the model to analyze. Prompt design can strongly affect the responses returned by the LLM.

![Output after running the Spark stored procedure][6]

## Benefits of the integration

1. This integration enables you to use MongoDB for OLTP and BigQuery for OLAP, providing a complete data management solution.
2. Once the data is transformed and copied to BigQuery, BigQuery ML lets you [create and run machine learning (ML) models](https://cloud.google.com/bigquery/docs/e2e-journey) by using GoogleSQL queries. 
3. BigQuery ML also lets you [access LLMs and Cloud AI APIs](https://cloud.google.com/bigquery/docs/ai-application-overview) to perform artificial intelligence (AI) tasks like text generation and machine translation. 

## Conclusion

By combining the power of BigQuery, Spark stored procedures, and MongoDB, you can create a robust and scalable data processing pipeline that leverages the strengths of each technology. BigQuery provides a reliable and scalable data warehouse for storing and analyzing structured data, while Spark allows you to process and transform data from various sources, including semi-structured and unstructured data from MongoDB. Spark stored procedures enable you to encapsulate and reuse this logic, making it easier to maintain and optimize your data processing workflows.

### Further reading

1. [Get started](https://www.mongodb.com/docs/atlas/getting-started/) with MongoDB Atlas on Google Cloud.
2. Work with [stored procedures](https://cloud.google.com/bigquery/docs/spark-procedures) for Apache Spark.
3. [Create](https://cloud.google.com/bigquery/docs/create-machine-learning-model) machine learning models in BigQuery ML.


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt3cc99ff9b6ad9cec/66155da90c478454e8a349f1/1.png
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltdd635dea2d750e73/66155dc254d7c1521e8eea3a/2.png
  [3]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blta48182269e87f9e7/66155dd2cbc2fbae6d8175ea/3.png
  [4]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltb399b53e83efffb9/66155de5be36f52825d96ea5/4.png
  [5]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt6efef490b3d34cf0/66155dfd2b98e91579101401/5.png
  [6]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt56b91d83e11dea5f/66155e0f7cacdc153bd4a78b/6.png

Spark Up Your MongoDB and BigQuery using BigQuery Spark Stored Procedures

Yemi Falokun

ISV CoE Solution Architect at Google

In the digital age, leveraging natural language for database queries represents a leap toward more intuitive data management. [Vertex AI Extensions](https://cloud.google.com/vertex-ai/generative-ai/docs/extensions/overview), currently in **private preview**, help in interacting with MongoDB using natural language. This tutorial introduces an approach that combines Google [Gemini's](https://ai.google.dev/models/gemini) advanced natural language processing with MongoDB, facilitated by Vertex AI Extensions. These extensions address key limitations of large language models (LLMs) by enabling real-time data querying and modification, which LLMs cannot do due to their static knowledge base post-training. By integrating MongoDB Atlas with Vertex AI Extensions, we offer a solution that enhances the accessibility and usability of the database. 

MongoDB's dynamic schema, scalability, and comprehensive querying capabilities render it exemplary for Generative AI applications. It is adept at handling the versatile and unpredictable nature of data that these applications generate and use. From personalized content generation, where user data shapes content in real time, to sophisticated, AI-driven recommendation systems leveraging up-to-the-minute data for tailored suggestions, MongoDB stands out. Furthermore, it excels in complex data analysis, allowing AI tools to interact with vast and varied datasets to extract meaningful insights, showcasing its pivotal role in enhancing the efficiency and effectiveness of Generative AI applications.

## Natural language to MongoDB queries

Natural language querying represents a paradigm shift in data interaction, allowing users to retrieve information without the need for custom query languages. By integrating MongoDB with a system capable of understanding and processing natural language, we streamline database operations, making them more accessible to non-technical users.

### Solution blueprint

The solution involves a synergy of several components, including MongoDB, the Google Vertex AI SDK, Google Secrets Manager, and OpenAPI 3 specifications. Together, these elements create a robust framework that translates natural language queries into MongoDB Data API calls. In this solution, we have explored basic CRUD operations with Vertex AI Extensions. We are closely working with Google to enable vector search aggregations in the near future.

### Components involved

1. **MongoDB**: A versatile, document-oriented database that stores data in JSON-like formats, making it highly adaptable to various data types and structures
2. **Google Vertex AI SDK**: Facilitates the creation and management of AI and machine learning models, including the custom extension for Google Vertex AI 
3. **Vertex AI Extensions:** Enhance LLMs by allowing them to interact with external systems in real-time, extending their capabilities beyond static knowledge 
4. **Google Secrets Manager**: Securely stores sensitive information, such as MongoDB API keys, ensuring the solution's security and integrity
5. **OpenAPI 3 Specification for MongoDB Data API**: Defines a standard, language-agnostic interface to MongoDB that allows for both easy integration and clear documentation of the API's capabilities

### Description of the solution

The solution operates by converting natural language queries into parameters that the MongoDB Data API can understand. This conversion is facilitated by a custom extension developed using the Google Vertex AI extension SDK, which is then integrated with Gemini 1.0 Pro. The extension leverages OpenAPI 3 specifications to interact with MongoDB, retrieving data based on the user's natural language input. Google Secrets Manager plays a critical role in securely managing API keys required for MongoDB access, ensuring the solution's security.

![architecture diagram of the Vertex AI extensions][1]

## Prerequisites

Before you start, make sure you have the following:

1. Secure your access to the Google Cloud [project](https://console.cloud.google.com/) or to create a new project.
2. If you are new to MongoDB Atlas, you can sign up to MongoDB either through the [Google Cloud 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) or with the Atlas [registration page](https://account.mongodb.com/account/register).
3. Vertex AI Extensions are not publicly available. Please sign up for the Extensions [Trusted Tester Program](https://cloud.google.com/vertex-ai/generative-ai/docs/extensions/overview).
4. Basic knowledge of [OpenAPI specifications](https://swagger.io/specification/) and how to create them for APIs will be helpful.
5. You’ll need a Google [Cloud Storage bucket](https://cloud.google.com/storage/docs/creating-buckets) for storing the OpenAPI specifications.

Before we begin, also make sure you:

**Enable MongoDB Data API**: To enable the Data API from the Atlas console landing page, open the [Data API](https://www.mongodb.com/docs/atlas/app-services/data-api/generated-endpoints/) section from the side pane, enable the Data API, and copy the URL Endpoint as shown below.

![Navigating to Data API on MongoDB][2]

To create an API key, click on **Users** and then click on **Create API Key**.

![Creating API key for Data API][3]

**Create a secret for your API key**: To secure the MongoDB Data API key, store it in Google Cloud [Secret Manager](https://cloud.google.com/secret-manager/docs/creating-and-accessing-secrets#:~:text=Go to the Secret Manager page in the Google Cloud console.&text=On the Secret Manager page,example%2C my-secret )). To create a new secret on the Google Cloud Console, [navigate](https://console.cloud.google.com/security/secret-manager) to Secrets Manager, and click on **CREATE SECRET**. Paste the secret created from MongoDB to the secret value field and click on **Create**.

![Creating secret on Google Cloud console][4]

**Develop API specification:** We will define the interactions between Google Vertex AI and your MongoDB Data API using OpenAPI 3.0 specs. You can download the specs from the [GitHub repository](https://github.com/mongodb-partners/MongoDB-VertexAI-extensions). This specification outlines how natural language queries will be translated into MongoDB operations.

## Create Vertex AI extensions

This tutorial uses the MongoDB default dataset from the **sample_mflix** database, **movies** collection. We will run all the below code on the Enterprise Colab notebook.

1. Vertex AI Extensions is a platform for creating and managing extensions that connect large language models to external systems via APIs. These external systems can provide LLMs with real-time data and perform data processing actions on their behalf.

```python
from google.colab import auth
auth.authenticate_user("GCP project id")
!gcloud config set project {"GCP project id"}
```

2. Install the required Python dependencies.

```python
!gsutil cp gs://vertex_sdk_private_releases/llm_extension/google_cloud_aiplatform-1.44.dev20240315+llm.extension-py2.py3-none-any.whl .
!pip install --force-reinstall --quiet google_cloud_aiplatform-1.44.dev20240315+llm.extension-py2.py3-none-any.whl[extension]
!pip install --upgrade --quiet google-cloud-resource-manager
!pip install --force-reinstall --quiet langchain==0.0.298!pip install pytube
!pip install --upgrade google-auth
!pip install bigframes==0.26.0
```

3. Once the dependencies are installed, restart the kernel.

```python
import IPython
app = IPython.Application.instance()
app.kernel.do_shutdown(True)  # Re-run the Env variable cell again after Kernel restart
```

4. Initialize the environment variables.

```python
import os
## This is just a sample values please replace accordingly to your project
# Setting up the GCP project
os.environ['PROJECT_ID'] = 'gcp project id'  # GCP Project ID
os.environ['REGION'] =  "us-central1" # Project Region
## GCS Bucket location
os.environ['STAGING_BUCKET'] =  "gs://vertexai_extensions"
## Extension Config
os.environ['EXTENSION_DISPLAY_HOME'] =  "MongoDb Vertex API Interpreter"
os.environ['EXTENSION_DESCRIPTION'] =  "This extension makes api call to mongodb to do all crud operations"


## OPEN API SPec config
os.environ['MANIFEST_NAME'] =  "mdb_crud_interpreter"
os.environ['MANIFEST_DESCRIPTION'] =  "This extension makes api call to mongodb to do all crud operations"
os.environ['OPENAPI_GCS_URI'] =  "gs://vertexai_extensions/mongodbopenapispec.yaml"


## API KEY secret location
os.environ['API_SECRET_LOCATION'] = "projects/787220387490/secrets/mdbapikey/versions/1"


##LLM config
os.environ['LLM_MODEL'] = "gemini-1.0-pro"
```

5. Download the Open API specification from [GitHub](https://github.com/mongodb-partners/MongoDB-VertexAI-extensions) and upload the [YAML file](https://github.com/mongodb-partners/MongoDB-VertexAI-extensions/blob/main/open-api-spec/mdb-data-api.yaml) to the Google Cloud Storage bucket. 

```python
from google.cloud import aiplatformfrom google.cloud.aiplatform.private_preview import llm_extension

PROJECT_ID = os.environ['PROJECT_ID']
REGION = os.environ['REGION']
STAGING_BUCKET = os.environ['STAGING_BUCKET']

aiplatform.init(
    project=PROJECT_ID,
    location=REGION,
    staging_bucket=STAGING_BUCKET,
)
```

6. To create the Vertex AI extension, run the below script. The manifest here is a structured JSON object containing several key components:

```python
mdb_crud = llm_extension.Extension.create(
display_name = os.environ['EXTENSION_DISPLAY_HOME'],
description = os.environ['EXTENSION_DESCRIPTION'],  # Optional         manifest = {             "name": os.environ['MANIFEST_NAME'],
             "description": os.environ['MANIFEST_DESCRIPTION'],
             "api_spec": {
                 "open_api_gcs_uri": os.environ['OPENAPI_GCS_URI'],
             },             "auth_config": {
                 "apiKeyConfig":{
             "name":"api-key",
            "apiKeySecret":os.environ['API_SECRET_LOCATION'],
            "httpElementLocation": "HTTP_IN_HEADER"
            },
            "authType":"API_KEY_AUTH"
            },
        },
)
```

7. Validate the Created Extension, and print the Operation Schema and Parameters.

```python
print("Name:", mdb_crud.gca_resource.name)print("Display Name:", mdb_crud.gca_resource.display_name)print("Description:", mdb_crud.gca_resource.description)
import pprint
pprint.pprint(mdb_crud.operation_schemas())
```

## Extension in action

Once the extension is created, navigate to [Vertex AI UI](https://console.cloud.google.com/vertex-a) and then Vertex UI Extension on the left pane. 

![Navigating to Vertex AI extensions][5]

All the created extensions will be available for selection. Select **MongoDB Vertex API Interpreter**. For this tutorial, we will consider two user journeys.

### User Journey 1

**Retrieve data through Vertex AI extensions:** Imagine needing to quickly find out when a classic film was released without navigating through the database manually. By asking Vertex AI, "Find the release year of the movie 'A Corner in Wheat' from VertexAI-POC cluster, sample_mflix, movies," you get the specific release year, as the system performs a findOne() operation to retrieve this detail.

![Testing Vertex AI extensions for find query][6]

In addition to using the UI, we can also use the Notebook to query MongoDB using natural language processing. We are using Gemini to convert natural language to the actual operation schema for calling the Vertex AI extension. Update the environment variable and run the below script.

```python
## Please replace accordingly to your project
## Operation Ids
os.environ['FIND_ONE_OP_ID'] = "findone_mdb"




## NL Queries
os.environ['FIND_ONE_NL_QUERY'] = "Find the release year of the movie 'A Corner in Wheat' from VertexAI-POC cluster, sample_mflix, movies"


## Mongodb Config
os.environ['DATA_SOURCE'] = "VertexAI-POC"
os.environ['DB_NAME'] = "sample_mflix"
os.environ['COLLECTION_NAME'] = "movies"


### Test data setup
os.environ['TITLE_FILTER_CLAUSE'] = "A Corner in Wheat"

from vertexai.preview.generative_models import GenerativeModel, Tool

fc_chat = GenerativeModel(os.environ['LLM_MODEL']).start_chat()
findOneResponse = fc_chat.send_message(os.environ['FIND_ONE_NL_QUERY'],
    tools=[Tool.from_dict({        "function_declarations": mdb_crud.operation_schemas()
   })],
)

findOneResponse
```

From the converted schema, query the extension.

```python
response = mdb_crud.execute(
    operation_id = findOneResponse.candidates[0].content.parts[0].function_call.name,
    operation_params = findOneResponse.candidates[0].content.parts[0].function_call.args
)

response
```

### User Journey 2

**Retrieve data through Vertex AI extensions:** A film historian wants a list of all movies released in a specific year — say 1924 — to study the cinematic trends of that era. They could ask, "Give me movies released in the year 1924 from VertexAI-POC cluster, sample_mflix, movies," and the system would use the find() method to list all movies from 1924, providing a comprehensive snapshot of that year's cinematic output.

![testing Vertex AI extensions for find query 2][7]

In addition to using the UI, we can also use the Notebook to query MongoDB using natural language processing. 

Here’s the Gemini integration for natural language to extension schema conversion:

```python
## This is just a sample values please replace accordingly to your project
## Operation Ids
os.environ['FIND_MANY_OP_ID'] = "findmany_mdb"


## NL Queries
os.environ['FIND_MANY_NL_QUERY'] = "give me movies released in year 1924 from VertexAI-POC cluster, sample_mflix, movies"




## Mongodb Config
os.environ['DATA_SOURCE'] = "VertexAI-POC"
os.environ['DB_NAME'] = "sample_mflix"
os.environ['COLLECTION_NAME'] = "movies"
os.environ['YEAR'] = 1924
fc_chat = GenerativeModel(os.environ['LLM_MODEL']).start_chat()
findmanyResponse = fc_chat.send_message(os.environ['FIND_MANY_NL_QUERY'],
    tools=[Tool.from_dict({        "function_declarations": mdb_crud.operation_schemas()
   })],
)

findmanyResponse
```

Execute the extension to get the response.

```python
response = mdb_crud.execute(
    operation_id = findmanyResponse.candidates[0].content.parts[0].function_call.name,
    operation_params = findmanyResponse.candidates[0].content.parts[0].function_call.args
)

response
```

## Step by step extension execution

![User queries with natural language][8]

![LLM converts natural language accordingly to extension schema and executes extension][9]

![Extension fetches the document from MongoDB and returns the response][10]

![LLM converts the extension response to natural language and responds back to user][11]

## Conclusion

In conclusion, integrating MongoDB with natural language querying capabilities revolutionizes data interaction, enhancing accessibility and intuitiveness for database queries. Leveraging the Google Gemini Foundation Model alongside a custom Vertex AI extension not only enriches the data retrieval experience but also upholds data security and integrity. We are closely working with the Google Cloud team to add even more query patterns to Vertex AI extensions. Watch out for more in this space. 

### Further reading

1. [Get started](https://www.mongodb.com/docs/atlas/getting-started/) with MongoDB Atlas on Google Cloud.
2. Connect models to APIs by using [Vertex AI extensions](https://cloud.google.com/vertex-ai/generative-ai/docs/extensions/overview).


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt5f6e6aa6cea13ba1/661471b70c47840e25a3437a/1.png
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltae31621998903a57/661471cd4180c1c4ede408cb/2.png
  [3]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt58474a722f262f1a/661471e40d99455ada032667/3.png
  [4]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt9cd6a0e4c6b2ed4c/661471f5da0c3a5c7ff77441/4.png
  [5]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt3ac5e7c88ed9d678/661472114180c1f08ee408d1/5.png
  [6]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt39e9b0f8b7040dab/661472241a0e49338babc9e1/6.png
  [7]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltfc26acb17bfca16d/6614723b2b98e9f356100e6b/7.png
  [8]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt7bdaa1e8a1cf5a51/661472517cacdc0fbad4a075/8.png
  [9]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt272144a86fea7776/661472632b98e9562f100e6f/9.png
  [10]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt62198b1ba0785a55/66147270be36f54af2d96927/10.png
  [11]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt4a7e5371abe658e1/66147281be36f5ed61d9692b/11.png

Harnessing Natural Language for MongoDB Queries with Google Gemini

Stanimira Vlaeva

Developer Advocate

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

Diana Annie Jenosh

Senior Partner Solutions Architect

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 today's data-driven world, the ability to analyze and efficiently move data across different platforms is crucial. MongoDB Atlas and Google BigQuery are two powerful platforms frequently used for managing and analyzing data. While they excel in their respective domains, connecting and transferring data between them seamlessly can pose challenges. However, with the right tools and techniques, this integration becomes not only possible but also streamlined.

One effective way to establish a smooth pipeline between MongoDB Atlas and BigQuery is by leveraging PyMongoArrow and pandas-gbq, two powerful Python libraries that facilitate data transfer and manipulation. PyMongoArrow acts as a bridge between MongoDB and Arrow, a columnar in-memory analytics layer, enabling efficient data conversion. On the other hand, pandas-gbq is a Python client library for Google BigQuery, allowing easy interaction with BigQuery datasets.

![Image 1: Architecture diagram for MongoDB interaction with BigQuery and VertexAI using python libraries][1]

## Advantages of a Python-based solution

1. Easily implement movement of a wide range of datatypes between MongoDB and BigQuery.
1. Easily join multiple data sources like cloud storage, Google databases, MongoDB Atlas etc. and transform the data using pandas dataframes.
1. You can use your favorite notebook to build the solution, including the new preview notebook available in BigQuery Studio.
1. Perform [exploratory data analysis](https://en.wikipedia.org/wiki/Exploratory_data_analysis) on data read from both Google BigQuery and MongoDB Atlas platforms without physically moving the data between these platforms. This will simplify the effort required by data engineers to move the data and offers a faster way for data scientists to build machine learning (ML) models.

Let's discuss each of the implementation advantages with examples.

### ETL data from MongoDB to BigQuery

Let’s consider a [sample shipwreck dataset](https://www.mongodb.com/docs/atlas/sample-data/) available on MongoDB Atlas for this use case.

Use the commands below to install the required libraries on the notebook environment of your choice. For easy and scalable setup, use [BigQuery Jupyter notebooks](https://cloud.google.com/bigquery/docs/visualize-jupyter#query_notebook_data_using_the_bigquery_magic_command) or managed VertexAI [workbench](https://console.cloud.google.com/vertex-ai/workbench/managed) notebooks.

![Image 2: screenshot of jupyter notebook for BigQuery DataFrames implementation.][2]

```
!pip install --upgrade pip
!pip install pymongoarrow
!pip install pandas-gbq
```

First, establish a connection to your MongoDB Atlas cluster using PyMongoArrow. This involves configuring authentication and selecting the database and collection from which you want to transfer data. Follow MongoDB Atlas [documentation](https://www.mongodb.com/docs/atlas/getting-started/) for setting up your cluster, network access, and authentication. [Load a sample dataset](https://www.mongodb.com/docs/atlas/sample-data/) to your Atlas cluster. [Get the Atlas connection string](https://www.mongodb.com/docs/atlas/tutorial/connect-to-your-cluster/) and replace the URI string below with your connection string. The below script is also available in the [GitHub repository](https://github.com/mongodb-developer/MongoDB_to_BigQuery_using_PyMongoArrow) with steps to set up.

```python
#Read data from MongoDB
import certifi
import pprint 
import pymongo
import pymongoarrow 
from pymongo import MongoClient

client = MongoClient("URI ``sting``",tlsCAFile=certifi.where())

#Initialize database and collection
db = client.get_database("sample_geospatial")
col = db.get_collection("shipwrecks")

for doc in col.find({}): 
  pprint.pprint(doc)

from pymongoarrow.monkey import patch_all 
patch_all()

#Create Dataframe for data read from MongoDB
import pandas as pd 
df = col.find_pandas_all({})
```

Transform the data to the required format — e.g., transform and remove the unsupported data formats, like the MongoDB object ID, or convert the MongoDB object to JSON before writing it to BigQuery. Please refer to the documentation to learn more about [data types](https://arrow.apache.org/docs/cpp/api/datatype.html) supported by pandas-gbq and [PyMongoArrow](https://mongo-arrow.readthedocs.io/en/latest/data_types.html).

```python
#Transform the schema for required format. 
#e.g. the object id is not supported in dataframes can be removed or converted to string. 

del(df["_id"])
```

Once you have retrieved data from MongoDB Atlas and converted it into a suitable format using PyMongoArrow, you can proceed to transfer it to BigQuery using either the pandas-gbq or google-cloud-bigquery. In this article, we are using pandas-gbq. Refer to the [documentation](https://cloud.google.com/bigquery/docs/pandas-gbq-migration) for more details on the differences between pandas-gbq and google-cloud-bigquery libraries. Ensure you have a dataset in BigQuery to which you want to load the MongoDB data. You can create a new dataset or use an existing one.

```python
#Write the transformed data to BigQuery.

import pandas_gbq

pandas_gbq.to_gbq(df[0:100], "gcp-project-name.bigquery-dataset-name.bigquery-table-name", project_id="gcp-project-name")
```

As you embark on building your pipeline, optimizing the data transfer process between MongoDB Atlas and BigQuery is essential for performance. A few points to consider:

1. Batch Dataframes in to chunks, especially when dealing with large datasets, to prevent memory issues.
1. Handle schema mapping and data type conversions properly to ensure compatibility between the source and destination databases.
1. With the right tools like Google colab, VertexAI Workbench etc, this pipeline can become a cornerstone of your data ecosystem, facilitating smooth and reliable data movement between MongoDB Atlas and Google BigQuery.

### Introduction to Google BigQuery DataFrames (bigframes)

Google bigframes is a Python API that provides a pandas-compatible DataFrame and machine learning capabilities powered by the BigQuery engine. It provides a familiar pandas interface for data manipulation and analysis. Once the data from MongoDB is written into BigQuery, the BigQuery DataFrames can unlock the user-friendly solution for analyzing petabytes of data with ease. The pandas DataFrame can be read directly into BigQuery DataFrames using the Python bigframes.pandas library. Install the bigframes library to use BigQuery DataFrames.

```
!pip install bigframes
```

Before reading the pandas DataFrames into BigQuery DataFrames, rename the columns as per Google's [schema guidelines](https://cloud.google.com/bigquery/docs/schemas#column_names). (Please note that at the time of publication, the feature may not be GA).

```python
import bigframes.pandas as bpd
bigframes.options.bigquery.project = "GCP project ID"

# df = <pandas_df>
bdf = bpd.read_pandas(df)
```

For more information on using Google Cloud Bigquery DataFrames, visit the Google Cloud [documentation](https://cloud.google.com/python/docs/reference/bigframes/latest).

## Conclusion

Creating a robust pipeline between MongoDB Atlas and BigQuery using PyMongoArrow and pandas-gbq opens up a world of possibilities for efficient data movement and analysis. This integration allows for the seamless transfer of data, enabling organizations to leverage the strengths of both platforms for comprehensive data analytics and decision-making.

### Further reading

- Learn more about MongoDB [PyMongoArrow](https://mongo-arrow.readthedocs.io/en/latest/quickstart.html) libraries and how to use them.
- Read more about Google [BigQuery DataFrames](https://cloud.google.com/python/docs/reference/bigframes/latest) for pandas and ML.
- Load DataFrames to BigQuery with Google [pandas-gbq](https://cloud.google.com/bigquery/docs/samples/bigquery-pandas-gbq-to-gbq-simple).


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt2557413e5cba18f3/65c3cbd20872227d14497236/image1.png
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt7459376aeef23b01/65c3cbd2245ed9a8b190fd38/image2.png

Orchestrating MongoDB & BigQuery for ML Excellence with PyMongoArrow and BigQuery Pandas Librarie

Orchestrating MongoDB & BigQuery for ML Excellence with PyMongoArrow and BigQuery Pandas Libraries

Ashwin Gangadhar

The application development landscape is evolving very rapidly. Today, users crave intuitive, context-aware experiences that understand their intent and deliver relevant results even when queries aren't perfectly phrased, putting an end to the keyword-based search practices. This is where MongoDB Atlas and Google Cloud Vertex AI can help users build and deploy scalable and resilient applications. 

[MongoDB Atlas Vector Search](https://www.mongodb.com/products/platform/atlas-vector-search) is a cutting-edge tool that indexes and stores high-dimensional vectors, representing the essence of your data. It allows you to perform lightning-fast similarity searches, retrieving results based on meaning and context. Google Vertex AI is a comprehensive AI platform that houses an abundance of pre-trained models and tools, including the powerful [Vertex AI](https://cloud.google.com/vertex-ai/docs) PALM. This language model excels at extracting semantic representations from text data, generating those crucial vectors that fuel MongoDB Atlas Vector Search. 

Vector Search can be useful in a variety of contexts, such as natural language processing and recommendation systems. It is a powerful technique that can be used to find similar data based on its meaning.

In this tutorial, we will see how to get started with MongoDB Atlas and Vertex AI. If you are new to MongoDB Atlas, refer to the [documentation](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) to get set up from Google Cloud Marketplace or use the [Atlas registration page](https://account.mongodb.com/account/register). 

## Before we begin

Make sure that you have the below prerequisites set up before starting to test your application.

1. MongoDB Atlas access, either by the [registration page](https://account.mongodb.com/account/register) or from [Google Cloud Marketplace](https://console.cloud.google.com/marketplace/product/mongodb/mdb-atlas-self-service?project=mdb-gcp-marketplace&utm_source=nurture+&utm_medium=email+&utm_campaign=Int_EM_GC+drip+series+dev+1_12_23_WW&utm_id=mongodb&elqTrackId=A66D8FD3837BB72F4DBB8CABAC0A9F37&elqTrack=true)
2. Access to [Google Cloud Project](https://console.cloud.google.com/welcome) to deploy and create a [Compute Engine instance](https://console.cloud.google.com/compute/instances)

## How to get set up

Let us consider a use case where we are loading sample PDF documents to MongoDB Atlas as vectors and deploying an application on Google Cloud to perform a vector search on the PDF documents. 

We will start with the creation of MongoDB Atlas Vector Search Index on the collection to store and retrieve the vectors generated by the Google Vertex AI PALM model. To store and access vectors on MongoDB Atlas, we need to create an [Atlas Search index](https://www.mongodb.com/docs/atlas/atlas-vector-search/vector-search-tutorial/?lb-height=100%&lb-width=100%&presentation=true#prerequisites). 

### Create an Atlas Search index 

1. Navigate to the **Database Deployments** page for your project.
2. Click on **Create Database.** Name your Database **vertexaiApp** and your collection **chat-vec**.
3. Click **Atlas Search** from the Services menu in the navigation bar.
4. Click **Create Search Index** and select **JSON Editor** under **Atlas Vector Search**. Then, click **Next.**
5. In the **Database and Collection** section, find the database **vertexaiApp**, and select the **chat-vec** collection.
6. Replace the default definition with the following index definition and then click **Next**. Click on **Create Search index** on the review page.

```json
{
    "fields": [
        {
            "type":"vector",
            "path":"vec",
            "numDimensions":768,
            "similarity": "cosine"
        }
    ]
}
```

### Create a Google Cloud Compute instance

We will create a [Google Cloud virtual machine instance](https://cloud.google.com/compute/docs/instances/create-start-instance#console) to run and deploy the application. The Google Cloud VM can have all the default configurations. To begin, log into your Google Cloud Console and perform the following steps:

- In the Google Cloud console, click on **Navigation menu > Compute Engine.**
- Create a new VM instance with the below configurations:
    - **Name:** vertexai-chatapp
    - **Region**: region near your physical location
    - **Machine configurations:**
        - Machine type: High Memory, n1-standard-1
- Boot disk: Click on **CHANGE**
    - Increase the size to 100 GB.
    - Leave the other options to default (Debian).
- Access: Select **Allow full access** to all Cloud APIs.
- Firewall: Select all.
- Advanced options:
    - Networking: Expand the default network interface.
    - For External IP range: Expand the section and click on **RESERVE STATIC EXTERNAL IP ADDRESS**. This will help users to access the deployed application from the internet.
    - Name your IP and click on **Done**.
- Click on **CREATE** and the VM will be created in about two to three minutes.

### Deploy the application

Once the VM instance is created, SSH into the VM instance and clone the [GitHub repository](https://github.com/mongodb-partners/MongoDB-VertexAI-Qwiklab.git).

```
git clone https://github.com/mongodb-partners/MongoDB-VertexAI-Qwiklab.git
```

The repository contains a script to create and deploy a [Streamlit](https://streamlit.io/) application to transform and store PDFs in MongoDB Atlas, then search them lightning-fast with Atlas Vector Search. The app.py script in the repository uses Python and LangChain to leverage MongoDB Atlas as our data source and Google Vertex AI for generating embeddings. 

We start by setting up connections and then utilize LangChain’s ChatVertexAI and Google's Vertex AI embeddings to transform the PDF being loaded into searchable vectors. Finally, we constructed the Streamlit app structure, enabling users to input queries and view the top retrieved documents based on vector similarity.

Install the required dependencies on your virtual machine using the below commands:

```bash
sudo apt update
sudo apt install python3-pip
sudo apt install git
git --version
pip3 --version
cd MongoDB-VertexAI-Qwiklab
pip3 install -r requirements.txt
```

Once the requirements are installed, you can run the application using the below command. Open the application using the public IP of your VM and the port mentioned in the command output:

```bash
streamlit run app.py
```

![Screenshot of the application for uploading the documents][1]

The application uses the *PdfReader* library to read the PDF file from the user and chunks the data into batches. Each chunk is converted into vectors using LangChain’s *VertexAIEmbeddings* libraries. The uploaded PDF is written into MongoDB Atlas in chunks, along with the vectors. The data can be viewed in vertexaiApp.chat-vec namespace documents.

![Screenshot of the vectors being stored into MongoDB Atlas][2]

To perform a search on the data, navigate to the Q&A tab on your application. Type in your query and press enter. The application will perform a vector search on MongoDB documents using the index created earlier. The application uses the chat-bison model to converse with the user.

![Screenshot of the application for searching the PDF data from MongoDB using Atlas Vector Search][3]

## Conclusion

In this tutorial, you learned how to build a simple application using MongoDB Atlas and Google Cloud Vertex AI using LangChain to access MongoDB Atlas. Also, we have seen how to deploy the application on Google Cloud Compute instance and access it using the public static IP address allocated. 

Try MongoDB Atlas for free in [Google Cloud Marketplace](https://console.cloud.google.com/marketplace/product/mongodb/mdb-atlas-self-service?project=mdb-gcp-marketplace&utm_source=nurture+&utm_medium=email+&utm_campaign=Int_EM_GC+drip+series+dev+1_12_23_WW&utm_id=mongodb&elqTrackId=A66D8FD3837BB72F4DBB8CABAC0A9F37&elqTrack=true) using our pay-as-you-go model and take advantage of our simplified billing.


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blted03916b8de19681/65a7ede75d51352518fb89d6/1.png
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltacbaf40c14ecccbc/65a7ee01cdbb961f8ac4faa4/2.png
  [3]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltc7a3b3c7a2db880c/65a7ee187a1dd7b984e136ba/3.png

Learn how to leverage MongoDB Atlas Vector Search to perform semantic search, Google Vertex AI for AI capabilities, and LangChain for seamless integration to build smart applications.

Build Smart Applications With Atlas Vector Search and Google Vertex AI

Ozan Güzeldereli

Senior Partner Solutions Engineer

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

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.