Venkatesh Shanbhag

Partner Solution Architect

Partner

Maruti C

Global Partner Engineering Lead, Google

Imagine a world of data-driven applications, demanding flexibility and power. This is where MongoDB thrives, with features perfectly aligned with these modern needs. But data alone isn't enough. Applications need intelligence too. Enter generative AI (gen AI), a powerful tool for content creation. But what if gen AI could do even more?


This is where AI agents come in. Acting as the mastermind behind gen AI, they orchestrate tasks, learn continuously, and make decisions. With agents, gen AI transforms into a versatile tool, automating tasks, personalizing interactions, and constantly improving. But how do we unleash this full potential?


Here's where the Vertex AI [Reasoning Engine](https://cloud.google.com/vertex-ai/generative-ai/docs/reasoning-engine/overview) steps in. Reasoning Engine (LangChain on Vertex AI) is a managed service that helps you to build and deploy an agent reasoning framework. It is a platform specifically designed for intelligent gen AI applications. Reasoning Engine is a Vertex AI service that has all the benefits of Vertex AI integration: security, privacy, observability, and scalability. Easily deploy and scale your application from development to production with a straightforward API, minimizing time-to-market. As a managed service, Reasoning Engine empowers you to build and deploy agent reasoning framework. It offers flexibility in how much reasoning you delegate to the large language model (LLM) and how much you control with custom code.


![Figure 1 : How it works: MongoDB as vector store for Google Reasoning engine][1]

Figure 1 : How it works: MongoDB as vector store for Google Reasoning engine

Lets see how MongoDB Atlas and Vertex AI Reasoning Engine can help you build and deploy a new generation of intelligent applications using LangChain on Vertex AI by combining data, automation, and machine learning. Here's a breakdown of the benefits: \

1. **Powerful and flexible data management with MongoDB:** MongoDB's features like data store and vector store are suited for modern data-driven applications that require flexibility and scalability.
2. **Enhanced applications with generative AI:** Generative AI can create content, potentially saving time and resources.
3. **Intelligent workflows with AI agents:** AI agents can manage and automate tasks behind the scenes, improving efficiency. They can learn from data and experience, constantly improving the application's performance. Agents can analyze data and make decisions, potentially leading to more intelligent application behavior. 


This solution is beneficial for various industries and applications, such as customer service chatbots that can learn and personalize interactions, or e-commerce platforms that can automate product recommendations based on customer data. Let's have a deep dive into the setup.


In this post, we will cover how to build a retrieval-augmented generation (RAG) application using MongoDB and Vertex AI and deploy it on Reasoning Engine. Firstly, we will ingest data into MongoDB Atlas and create embeddings for the RAG solution. We will also cover how to use agents to call different tools in return, querying different collections on MongoDB based on the context of the natural language query from the user. 



## Ingest data and vectors into MongoDB using LangChain

MongoDB Atlas simplifies the process by storing your complex data (like protein sequences or user profiles and so on) alongside their corresponding vector embeddings. This allows you to leverage vector search to efficiently find similar data points, uncovering hidden patterns and relationships.  Furthermore, MongoDB Atlas facilitates data exploration by enabling you to group similar data together based on their vector representations. 

LangChain is an open-source toolkit that helps developers build with LLMs. Like Lego for AI, it offers pre-built components to connect the models with your data and tasks. This simplifies building creative AI applications that answer questions, generate text formats, and more.

To begin with the setup, the first step is to create a [MongoDB Atlas cluster on Google Cloud](https://mongodb.prakticum-team.ru/products/platform/atlas-cloud-providers/google-cloud). Configure IP [access list entries](https://mongodb.prakticum-team.ru/docs/atlas/security/ip-access-list/) and a [database user](https://mongodb.prakticum-team.ru/docs/atlas/security-add-mongodb-users/) for accessing the cluster using the connection string. We will use [Google Colab](https://console.cloud.google.com/vertex-ai/colab/notebooks) to ingest, build, and deploy the RAG. 

Next, import the Python [notebook](https://github.com/mongodb-partners/MongoDB-VertexAI-Reasoning-Engine/blob/main/reasoning%20engine.ipynb) into your Colab enterprise, run the requirements, and ingest the block. We will import the data from Wikipedia for [Star Wars](https://en.wikipedia.org/wiki/Star_Wars) and [Star Trek](https://en.wikipedia.org/wiki/Star_Trek).

LangChain streamlines text embedding generation with pre-built models like text-embedding and textembedding-gecko. These models convert your text data into vector representations, capturing semantic meaning in a high-dimensional space. This facilitates efficient information retrieval and comparison within LangChain's reasoning workflows. We are using Google's text-embedding-004 model to convert the input data into embeddings on 768 dimensions. 

    def get_text_embeddings(chunks):
    from vertexai.language_models import TextEmbeddingModel
    model = TextEmbeddingModel.from_pretrained("text-embedding-004")
    inputs = chunks[0]
    embeddings = model.get_embeddings(chunks)
    return [embedding.values for embedding in embeddings]


The generated embeddings are stored in MongoDB Atlas alongside the actual data. Before executing the write_to_mongoDB function, update the URI to [connect](https://mongodb.prakticum-team.ru/docs/atlas/tutorial/connect-to-your-cluster/) to your MongoDB cluster. Pass the db_name and coll_name for the function where you want to store the embeddings.

    def write_to_mongoDB(embeddings, chunks, db_name, coll_name):
    from pymongo import MongoClient
    client = MongoClient("URI", tlsCAFile=certifi.where())
    db = client[db_name]
    collection = db[coll_name]
    
    for i in range(len(chunks)):
        collection.insert_one({
            "chunk": chunks[i],
            "embedding": embeddings[i]
        })


## Reasoning Engine


### Model

The first step in building your Reasoning Engine agent is specifying the generative AI model. Here, we're using the latest "gemini-1.5-pro" LLM, which will form the foundation of the RAG component.

    model = "gemini-1.5-pro-001"

## Tool creation: RAG using MongoDB Atlas with LangChain

LangChain acts as the bridge between your generative model and MongoDB Atlas, allowing it to query vectors. It takes a "query" as input, transforms it into embeddings using Google's [embedding models](https://cloud.google.com/vertex-ai/generative-ai/docs/embeddings/get-text-embeddings), and retrieves the most semantically near data from MongoDB Atlas. Below is the script for a tool that generates vectors for the query string, performs vector search on MongoDB Atlas, and returns the relevant document to the LLM. Update the function name database and collection name to read from different collections. We can initialize multiple tools and pass to the agent in the next step.

    def star_wars_query_tool(
    query: str) :
    """
    Retrieves vectors from a MongoDB database and uses them to answer a question related to Star wars.

    Args:
        query: The question to be answered about star wars.

    Returns:
        A dictionary containing the response to the question.
    """
    from langchain.chains import ConversationalRetrievalChain, RetrievalQA
    from langchain_mongodb import MongoDBAtlasVectorSearch
    from langchain_google_vertexai import VertexAIEmbeddings, ChatVertexAI
    from langchain.memory import ConversationBufferMemory, ConversationBufferWindowMemory
    from pymongo import MongoClient

    from langchain.prompts import PromptTemplate


    prompt_template = """Use the following pieces of context to answer the question at the end. If you don't know the answer, just say that you don't know, don't try to make up an answer. Do not return any answers from your own knowledge.
 
    {context}
    Question: {question}
    """
    # create prompt for LLM
    PROMPT = PromptTemplate(
        template=prompt_template, input_variables=["context", "question"]
    )

    # Add your connection string in srv format below in place of URI
    client = MongoClient("URI")
    db = client["embeddings"]

    embeddings =VertexAIEmbeddings(model_name="text-embedding-004")

    # initilize the vector store
    vs = MongoDBAtlasVectorSearch(
        collection=db["sample_starwars_embeddings"],
        embedding=embeddings,
        index_name="vector_index",
        embedding_key="embedding",
        text_key="chunk",
    )

    # initilize LLM
    llm = ChatVertexAI(
        model_name="gemini-1.5-pro",
        convert_system_message_to_human=True,
        max_output_tokens=1000,
    )

    # initilize retriver for the vector store object created
    retriever = vs.as_retriever(
        search_type="mmr", search_kwargs={"k": 10, "lambda_mult": 0.25}
    )
    memory = ConversationBufferWindowMemory(
        memory_key="chat_history", k=5, return_messages=True
    )

    # initilize the conversation chain
    conversation_chain = ConversationalRetrievalChain.from_llm(
        llm=llm,
        retriever=retriever,
        memory=memory,
        combine_docs_chain_kwargs={"prompt": PROMPT},
    )

    # query and get the response from conversation chain
    response = conversation_chain({"question": query})

    return response



## Define an agent

Vertex AI's Reasoning Engine agent goes beyond just decision-making tools, transforming LangChain agents into versatile AI assistants that can handle data, connect to systems, and make complex decisions, all while understanding and responding to text. This will let you tailor them to specific tasks like choosing the right tool for the job. Teaming up powerful language models like Gemini with reasoning agents enhances their skills by enabling them to understand and generate natural language, making them communication- and information-processing masters — a valuable addition to their toolkit.

By incorporating a reasoning layer, your agent leverages the provided tools to guide the end user toward achieving their ultimate objective. You can define multiple tools at the same time and the LLM will find out which tool to use based on the relevance to the question being asked and the description provided in the tool itself. We are using the default [LangchainAgent class](https://cloud.google.com/vertex-ai/generative-ai/docs/reference/python/latest/vertexai.preview.reasoning_engines.LangchainAgent) that can be further [customized](https://cloud.google.com/vertex-ai/generative-ai/docs/reasoning-engine/customize) based on your requirements. 

![Workflow for the above use case we discussed from end to end][2]


Figure 2: Workflow for the above use case we discussed from end to end

With the below code, we will initialize the agent for tools to perform vector search on MongoDB collections. The star_wars_query_tool will read from the sample_starwars_embeddings collection. Similarly, create a tool to read from the sample_startrek_embeddings collection. The Reasoning Engine will redirect the query to read from the Star Wars or Star Trek collection based on the reasoning and prompt set by the user while creating the tools.

    agent = reasoning_engines.LangchainAgent(
    model=model,
    tools=[star_wars_query_tool, star_trek_query_tool],
    agent_executor_kwargs={"return_intermediate_steps": True},
    ) 
    agent.query(input="tell me about star wars?")


## Deploy on Reasoning Engine

With the model, tools, and reasoning logic defined and tested locally, it's time to deploy your agent as a remote service on Vertex AI. We have used: 

    remote_agent = reasoning_engines.ReasoningEngine.create(
    agent,
    requirements=[
        "google-cloud-aiplatform[langchain,reasoningengine]",
        "cloudpickle==3.0.0",
        "pydantic==2.7.4",
        "langchain-mongodb",
        "pymongo",
        "langchain-google-vertexai",

    ],
    )


The output will include the deployment details for the Reasoning Engine that can be used to implement the user application.

    INFO:vertexai.reasoning_engines._reasoning_engines:reasoning_engine = vertexai.preview.reasoning_engines.ReasoningEngine('projects/project-id/locations/us-central1/reasoningEngines/reasoning-engine-id')

    from vertexai.preview import reasoning_engines
    REASONING_ENGINE_RESOURCE_NAME = "projects/project-id/locations/us-central1/reasoningEngines/reasoning-engine-id" 
    remote_agent = reasoning_engines.ReasoningEngine(REASONING_ENGINE_RESOURCE_NAME) 
    response = remote_agent.query(input="Tell me about episode 1 from wars")


You can also debug and optimize your agents by enabling tracing in the Reasoning Engine. View the [notebook](https://github.com/GoogleCloudPlatform/generative-ai/blob/main/gemini/reasoning-engine/tracing_agents_in_reasoning_engine.ipynb) that explains how you can use [Cloud Trace](https://cloud.google.com/trace) for exploring the tracing data to get insights.

Every aspect of your agent is customizable, from core instructions and starting prompts to managing conversation history for a seamless, context-aware experience across multiple queries. Follow the instructions in the Python [notebook](https://github.com/mongodb-partners/MongoDB-VertexAI-Reasoning-Engine/blob/main/reasoning%20engine.ipynb) of the [GitHub repository](https://github.com/mongodb-partners/MongoDB-VertexAI-Reasoning-Engine/tree/main) to create your own agent. The solution in this post can be easily extended to have an agent with multiple and any kind of [LangChain tools ](https://python.langchain.com/v0.1/docs/modules/model_io/chat/function_calling/) (like function calling and extensions) and to have an application with [multiple agents](https://blog.langchain.dev/langgraph-multi-agent-workflows/). We will talk about the multi-agents with MongoDB and Google Cloud in detail in our follow-up articles.

> Want $500 in credits for the Google Marketplace? Simply check out our [program](https://mongodb.prakticum-team.ru/lp/products/platform/atlas-cloud-providers/google-cloud/credit-offer?utm_campaign=gc_500offer_mp_listing&utm_source=google_cloud_mp&utm_medium=mp&utm_term=google), subscribe to Atlas, and claim your credits today, and try out [Atlas on the GCP marketplace](https://console.cloud.google.com/marketplace/product/mongodb/mdb-atlas-self-service?project=mdb-gcp-marketplace&utm_source=nurture%20&utm_medium=email%20&utm_campaign=Int_EM_GC%20drip%20series%20dev%201_12_23_WW&utm_id=mongodb&elqTrackId=A66D8FD3837BB72F4DBB8CABAC0A9F37&elqTrack=true) for your new workload.


  [1]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt9f64b82c56e27d9f/66be346a88b74e255fb12af2/ref_arch_1.png
  [2]: https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt1da2050faadd944f/66be3f7999b2ace6cd90df36/ref_arch_2.png

Build and deploy AI agents and RAG-based applications with MongoDB using Google's Vertex AI Reasoning Engine and LangChain.

Accelerate Your AI journey: Simplify Gen AI RAG With MongoDB Atlas & Google’s Vertex AI Reasoning Engine

Zi Wang

Partner Engineer at Google Cloud

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://mongodb.prakticum-team.ru/docs/atlas/getting-started/) a MongoDB Atlas cluster with [sample data](https://mongodb.prakticum-team.ru/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://mongodb.prakticum-team.ru/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

Explore how to extend BigQuery’s data processing to Apache Spark, and integrate MongoDB with BigQuery to facilitate data movement between the two.

Spark Up Your MongoDB and BigQuery Using BigQuery Spark Stored Procedures

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://mongodb.prakticum-team.ru/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://mongodb.prakticum-team.ru/docs/atlas/getting-started/) for setting up your cluster, network access, and authentication. [Load a sample dataset](https://mongodb.prakticum-team.ru/docs/atlas/sample-data/) to your Atlas cluster. [Get the Atlas connection string](https://mongodb.prakticum-team.ru/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

Maruti C

Content by Maruti C

Accelerate Your AI journey: Simplify Gen AI RAG With MongoDB Atlas & Google’s Vertex AI Reasoning Engine

Spark Up Your MongoDB and BigQuery Using BigQuery Spark Stored Procedures

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