Francesco Baldissera

Industry Solutions Consultant

MongoDB

Sebastian Rojas Arbulu

Specialist, Industry Solutions

Utilizing MongoDB Atlas for efficient data storage and Google Cloud Platform for complex calculations and hosting, businesses can implement real-time pricing adjustments based on customer behavior and competitor data.

Building a Dynamic Pricing Microservice with Vertex AI and MongoDB Atlas

Vittal Pai

Senior Solutions Architect

Ashwin Gangadhar

In the vast realm of Ecommerce, customers' ability to quickly and accurately search through an extensive range of products is paramount. [Atlas Vector Search](https://www.mongodb.com/products/platform/atlas-vector-search) is emerging as a turning point in this space, offering a refined approach to search that goes beyond mere keyword matching. Let's delve into its implementation using [MongoDB Atlas](https://www.mongodb.com/atlas/database), Atlas Vector Search, and [Databricks](https://www.databricks.com/).

### Prerequisites

* [MongoDB Atlas cluster](https://www.mongodb.com/docs/atlas/getting-started/)
* [Databricks cluster](https://docs.databricks.com/en/clusters/index.html)
* [python>=3.7](https://www.python.org/downloads/release/python-370/)
* [pip3](https://pypi.org/project/pip/)
* [Node.js and npm](https://docs.npmjs.com/downloading-and-installing-node-js-and-npm)
* [GitHub repo](https://github.com/mongodb-industry-solutions/retail-pricing-search-personaliztion) for AI-enhanced search and vector search (code is bundled up for clarity)

In a previous tutorial, [Learn to Build AI-Enhanced Retail Search Solutions with MongoDB and Databricks](https://www.mongodb.com/developer/products/atlas/retail-search-mongodb-databricks/), we showcased how the integration of MongoDB and Databricks provides a comprehensive solution for the retail industry by combining real-time data processing, workflow orchestration, machine learning, custom data functions, and advanced search capabilities as a way to optimize product catalog management and enhance customer interactions.

In this tutorial, we are going to be building the Vector Search solution on top of the codebase from the previous tutorial. Please check out the Github repository for the [full solution](https://github.com/mongodb-industry-solutions/retail-pricing-search-personaliztion/tree/main).

The diagram below represents the Databricks workflow for indexing data from the atp (available to promise), images, prd_desc (product discount), prd_score (product score), and price collections. These collections are also part of the previously mentioned tutorial, so please refer back if you need to access them.

Within the MongoDB Atlas platform, we can use change streams and the [MongoDB Connector for Spark](https://www.mongodb.com/docs/spark-connector/current/) to move data from the collections into a new collection called Catalog. From there, we will use a text transformer to create the **`Catalog Final Collection`**. This will enable us to create a corpus of indexed and vector embedded data that will be used later as the search dictionary. We’ll call this collection **`catalog_final_myn`**. This will be shown further along after we embed the product names.

The catalog final collection will include the available to promise status for each product, its images, the product discount, product relevance score, and price, along with the vectorized or embedded product name that we’ll point our vector search engine at.

With the image below, we explain what the [Databricks workflow](https://docs.databricks.com/en/workflows/index.html) looks like. It consists of two jobs that are separated in two notebooks respectively. We’ll go over each of the notebooks below.

![Overview of a Databricks job, including two pipelines to ingest data from MongoDB collections, transform that data, and vectorize it using a text transformer model](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/bltc3b2a88fc9350ffd/650db4f5c9f9dc0b117b978a/vectorsearch1.png)

## Indexing and merging several collections into one catalog

The first step is to ingest data from the previously mentioned collections using the spark.readStream method from the MongoDB Connector for Spark. The code below is part of the notebook we’ll set as a job using Databricks Workflows. You can learn more about [Databricks notebooks](https://docs.databricks.com/en/notebooks/index.html) by following their tutorial. 

```
atp = spark.readStream.format("mongodb").\ option('spark.mongodb.connection.uri', MONGO_CONN).\ option('spark.mongodb.database', "search").\ option('spark.mongodb.collection', "atp_status_myn").\ option('spark.mongodb.change.stream.publish.full.document.only','true').\ option('spark.mongodb.aggregation.pipeline',[]).\ option("forceDeleteTempCheckpointLocation", "true").load() atp = atp.drop("_id") atp.writeStream.format("mongodb").\ option('spark.mongodb.connection.uri', MONGO_CONN).\ option('spark.mongodb.database', "search").\ option('spark.mongodb.collection', "catalog_myn").\ option('spark.mongodb.operationType', "update").\ option('spark.mongodb.upsertDocument', True).\ option('spark.mongodb.idFieldList', "id").\ option("forceDeleteTempCheckpointLocation", "true").\ option("checkpointLocation", "/tmp/retail-atp-myn4/_checkpoint/").\ outputMode("append").\ start()
```
This part of the notebook reads data changes from the atp_status_myn collection in the search database, drops the _id field, and then writes (or updates) the processed data to the catalog_myn collection in the same database. 

Notice how it’s reading from the `atp_status_myn` collection, which already has the one hot encoding (boolean values if the product is available or not) from the [previous  tutorial](https://docs.google.com/document/d/1KbnO00DJrlZIXdWS_QCxfHQsbDB9jMm1z_5q44JYx78/edit?usp=sharing). This way, we make sure that we only embed the data from the products that are available in our stock.

Please refer to [the full notebook in our Github repository](https://github.com/mongodb-industry-solutions/retail-pricing-search-personaliztion/blob/main/notebooks/catalogIndexing.ipynb) if you want to learn more about all the data ingestion and transformations conducted during this stage. 

## Encoding text as vectors and building the final catalog collection

Using a combination of Python libraries and PySpark operations to process data from the Catalog MongoDB collection, we’ll transform it, vectorize it, and write the transformed data back to the Final Catalog collection. On top of this, we’ll build our application search business logic.

We start by using the %pip magic command, which is specific to Jupyter notebooks and IPython environments. The necessary packages are:
* **pymongo:** A Python driver for MongoDB.
* **tqdm:** A library to display progress bars.
* **sentence-transformers:** A library for state-of-the-art sentence, text, and image embeddings.

First, let’s use pip to install these packages in our Databricks notebook:

```
%pip install pymongo tqdm sentence-transformers
```

We continue the notebook with the following code: 

```
model = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
```

Here we load a [pre-trained model](https://huggingface.co/sentence-transformers/all-MiniLM-L6-v2) from the sentence-transformers library. This model will be used to convert text into embeddings or vectors. 

The next step is to bring the data from the MongoDB Atlas catalog and search collections. This as a continuation of the same notebook:
```
catalog_status = spark.readStream.format("mongodb").\
            option('spark.mongodb.connection.uri', MONGO_CONN).\
            option('spark.mongodb.database', "search").\
            option('spark.mongodb.collection', "catalog_myn").\
option('spark.mongodb.change.stream.publish.full.document.only','true').\
            option('spark.mongodb.aggregation.pipeline',[]).\
            option("forceDeleteTempCheckpointLocation", "true").load()
```
With this code, we set up a structured streaming read from the **`catalog_myn`** collection in the **`search`** database of MongoDB. The resulting data is stored in the **`catalog_status`** DataFrame in Spark. The read operation is configured to fetch the full document from MongoDB's change stream and does not apply any aggregation.

The notebook code continues with: 
```
#Calculating new column called discounted price using the F decorator

catalog_status = catalog_status.withColumn("discountedPrice", F.col("price") * F.col("pred_price"))

#One hot encoding of the atp_status column 

catalog_status = catalog_status.withColumn("atp", (F.col("atp").cast("boolean") & F.lit(1).cast("boolean")).cast("integer"))

#Running embeddings of the product titles with the get_vec function

catalog_status.withColumn("vec", get_vec("title"))

#Dropping _id column and creating a new final catalog collection with checkpointing

catalog_status = catalog_status.drop("_id")
catalog_status.writeStream.format("mongodb").\
            option('spark.mongodb.connection.uri', MONGO_CONN).\
            option('spark.mongodb.database', "search").\
            option('spark.mongodb.collection', "catalog_final_myn").\
            option('spark.mongodb.operationType', "update").\
            option('spark.mongodb.idFieldList', "id").\
            option("forceDeleteTempCheckpointLocation", "true").\
            option("checkpointLocation", "/tmp/retail-atp-myn5/_checkpoint/").\
            outputMode("append").\
            start()
```

With this last part of the code, we calculate a new column called discountedPrice as the product of the predicted price. Then, we perform [one-hot encoding](https://machinelearningmastery.com/why-one-hot-encode-data-in-machine-learning/) on the atp status column, vectorize the title of the product, and merge everything back into a final catalog collection.

Now that we have our catalog collection with its proper embeddings, it’s time for us to build the Vector Search Index using [MongoDB Atlas Search](https://www.mongodb.com/docs/atlas/atlas-search/). 

## Configuring the Atlas Vector Search index 

Here we’ll define how data should be stored and indexed for efficient searching. To configure the index, you can insert the snippet in MongoDB Atlas by browsing to your cluster splash page and clicking on the “Search” tab:
![Overview of the Search configuration panel for MongoDB Atlas collections](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt899d2c0d7b2a4268/650db8d9a0bdc093216ef75b/vectorsearch2.png)

Next, you can click over “Create Index.” Make sure you select “JSON Editor”:
![Overview of the JSON Editor functionality in the Search Configuration panel for MongoDB Atlas ](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt761abc22dd17f1fb/650db9239a515222b68bec9c/vectorsearch3.png)

Paste the JSON snippet from below into the JSON Editor. Make sure you select the correct database and collection! In our case, the collection name is **`catalog_final_myn`**. Please refer to the [full code in the repository](https://github.com/mongodb-industry-solutions/retail-pricing-search-personaliztion/blob/main/mdbindex/search.catalog_final_myn) to see how the full index looks and how you can bring it together with the rest of parameters for the [AI-enhanced search tutorial](https://www.mongodb.com/developer/products/atlas/retail-search-mongodb-databricks/).

```
{
  "mappings": {
    "fields": {
            "vec": [
        {
          "dimensions": 384,
          "similarity": "cosine",
          "type": "knnVector"
        }
      ]
    }
  }
}
```

In the code above, the vec field is of type [knnVector](https://www.mongodb.com/docs/atlas/atlas-search/field-types/knn-vector/#std-label-bson-data-types-knn-vector), designed for vector search. It indicates that each vector has 384 dimensions and uses cosine similarity to determine vector closeness. This is crucial for semantic search, where the goal is to find results that are contextually or semantically related.

By implementing these indexing parameters, we speed up retrieval times. Especially with high-dimensional vector data, as raw vectors can consume a significant amount of storage and reduce the computational cost of operations like similarity calculations. 

Instead of comparing a query vector with every vector in the dataset, indexing allows the system to compare with a subset, saving computational resources.

## A quick example of improved search results

[Browse over to our LEAFYY Ecommerce website](http://ec2-54-216-127-120.eu-west-1.compute.amazonaws.com:3000/), in which we will perform a search for the keywords ``tan bags``. You’ll get these results: 

![Search engine results page after running a simple query on an Ecommerce storefront powered with MongoDB Atlas Search](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt7a45a53f40d1e81e/650db9f4b258105f06802ddc/vectorsearch4.png)

As you can see, you’ll first get results that match the specific tokenized keywords “tan” and “bags”. As a result, this will give you any product that contains any or both of those keywords in the product catalog collection documents. 

However, not all the results are bags or of tan color. You can see shoes, wallets, a dress, and a pair of pants. This could be frustrating as a customer, prompting them to leave the site.

Now, enable vector search by clicking on the checkbox on the left of the magnifying glass icon in the search bar, and re-run the query “tan bags”. The results you get are in the image below: 
![Search engine results page after running a vector search query on an ecommerce storefront powered with MongoDB Atlas Search](https://images.contentstack.io/v3/assets/blt39790b633ee0d5a7/blt832782bc8dc0904a/650dba3427287d7d4dbf78de/vectorsearch5.png)

As you can see from the screenshot, the results became more relevant for a consumer. Our search engine is able to identify similar products by understanding the context that “beige” is a similar color to “tan”, and therefore showcase these products as alternatives.

## Conclusion

By working with MongoDB Atlas and Databricks, we can create real-time data transformation pipelines. We achieve this by leveraging the MongoDB Connector for Spark to prepare our operational data for vectorization, and store it back into our MongoDB Atlas collections. This approach allows us to develop the search logic for our Ecommerce app with minimal operational overhead.

On top of that, Atlas Vector Search provides a robust solution for implementing advanced search features, making it easy to deliver a great search user experience for your customers. By understanding and integrating these tools, developers can create search experiences that are fast, relevant, and user-friendly. 

Make sure to review the [full code in our GitHub repository](https://github.com/mongodb-industry-solutions/retail-pricing-search-personaliztion). [Contact us](https://www.mongodb.com/community/forums/) to get a deeper understanding of how to build advanced search solutions for your Ecommerce business. 


  

Learn how to implement Databricks Workflows and Atlas Vector Search for your Ecommerce accuracy.

How to Implement Databricks Workflows and Atlas Vector Search for Enhanced Ecommerce Search Accuracy

Learn how to utilize MongoDB and Databricks to build ai-enhanced retail search solutions.

Francesco Baldissera

Content by Francesco Baldissera

Building a Dynamic Pricing Microservice with Vertex AI and MongoDB Atlas

How to Implement Databricks Workflows and Atlas Vector Search for Enhanced Ecommerce Search Accuracy

Learn to Build AI-Enhanced Retail Search Solutions with MongoDB and Databricks