MongoDB Atlas via Spark

This notebook provides a top-level introduction in using Spark with MongoDB, enabling developers and data engineers to bring sophisticated real-time analytics and machine learning to live, operational data.

The following illustrates how to use MongoDB and Spark with an example application that leverages MongoDB's aggregation pipeline to pre-process data within MongoDB ready for use in Databricks. It shows as well how to query and write back to MongoDB for use in applications. This notebook covers:

1. How to read data from MongoDB into Spark.
2. How to run the MongoDB Connector for Spark as a library in Databricks.
3. How to leverage MongoDB's Aggregation Pipeline from within Spark
4. How to use the machine learning ALS library in Spark to generate a set of personalized movie recommendations for a given user.
5. How to write the results back to MongoDB so they are accessible to applications.

Create a Databricks Cluster and Add the Connector as a Library

1. Create a Databricks cluster.
2. Navigate to the cluster detail page and select the Libraries tab.
3. Click the Install New button.
4. Select Maven as the Library Source.
5. Use the Search Packages feature, find 'mongo-spark'. This should point to org.mongodb.spark:mongo-spark-connector_2.12:3.0.1 or newer.
6. Click Install.
   For more info on the MongoDB Spark connector (which now supports structured streaming) see the MongoDB documentation.

Create a MongoDB Atlas Instance

Atlas is a fully managed, cloud-based MongoDB service. We'll use Atlas to test the integration between MongoDb and Spark.

1. Sign up for MongoDB Atlas.
2. Create an Atlas free tier cluster.
3. Enable Databricks clusters to connect to the cluster by adding the external IP addresses for the Databricks cluster nodes to the whitelist in Atlas. For convenience you could (temporarily!!) 'allow access from anywhere', though we recommend to enable network peering for production.

Prep MongoDB with a sample data-set

MongoDB comes with a nice sample data-set that allows to quickly get started. We will use this in the context of this notebook

1. In MongoDB Atlas Load the sample data-set once the cluster is up and running.
2. You can confirm the presence of the data-set via the Browse Collections button in the Atlas UI.

Update Spark Configuration with the Atlas Connection String

1. Note the connect string under the Connect dialog in MongoDB Atlas. It has the form of "mongodb+srv://<username>:<password>@<databasename>.xxxxx.mongodb.net/"
2. Back in Databricks in your cluster configuration, under Advanced Options (bottom of page), paste the connection string for both the spark.mongodb.output.uri and spark.mongodb.input.uri variables. Please populate the username and password fields appropriately. This way all the workbooks you are running on the cluster will use this configuration.
3. Alternatively you can explicitly set the option when calling APIs like: spark.read.format("mongodb").option("spark.mongodb.input.uri", connectionString).load(). If you configured the variables in the cluster, you don't have to set the option.

Read data from the MongoDB 'sales' collection running an Aggregation Pipeline.

MongoDB's Aggregation Pipeline is a powerful capability that allows to pre-process and transform data within MongoDB. It's a great match for real-time analytics, dashboards, report generation with roll-ups, sums & averages with 'server-side' data post-processing. (Note: there is a whole book written about it).
MongoDB even supports rich secondary/compound indexes to extract, filter, and process only the data it needs – for example, analyzing all customers located in a specific geography right within the database without first having to load the full data-set, minimizing data-movement and reducing latency.
The below aggregation pipeline in our example has 4 stages:

1. Match stage : filters all documents which has "printer paper" in the items array.
   
2. Unwind stage : unwind the items array
   
3. Add fields stage : adds a new field called "totalSale" which is quantity of items sold * item price.
   
4. Project stage : only project "saleDate" and "totalSale" in the output

Read data from the MongoDB 'sales' collection as-is

For comparison we can read the data from the collection as-is without applying any aggregation pipeline transformation and show the schema.

Create a temp view

Let's use the dataframe created in above step and run a simple SparkSQL query on that

Writing Data to MongoDB to a new Collection

Now you can enrich the data with data coming from other sources, or use Spark MLLib for training ML Models using data in MongoDB.
For the demonstration here, we are simply writing the 'filtered_df' directly to a new MongoDB collection in the sample_supplies database.

View Data in MongoDB

You should be able to view the collection added in MongoDB Atlas in the Collections tab. Or you can use MongoDB Compass to view this via the Desktop app.

More Info

Relevant technologies that simplify real-time data flow and processing: Data Federation, Workload Isolation, Aggregation Pipelines, Materialized Views
For batch processing with Parquet on Object Store (S3) see the documentation on $out capabilities.