Anaiya Raisinghani

Data processing is no easy feat, but with the proper tools, it can be simplified and can enable you to make the best data-driven decisions possible. In a world overflowing with data, we need the best methods to derive the most useful information. 

The combination of [MongoDB Atlas](https://www.mongodb.com/atlas/database) with [Azure Databricks](https://learn.microsoft.com/en-us/azure/databricks/introduction/) makes an efficient choice for big data processing. By connecting Atlas with Azure Databricks, we can extract data from our Atlas cluster, process and analyze the data using [PySpark](https://spark.apache.org/docs/latest/api/python/), and then store the processed data back in our Atlas cluster. Using Azure Databricks to analyze your Atlas data allows for access to Databricks’ wide range of advanced analytics capabilities, which include machine learning, data science, and areas of artificial intelligence like natural language processing! Processing your Atlas data with these advanced Databricks tools allows us to be able to handle any amount of data in an efficient and scalable way, making it easier than ever to gain insights into our data sets and enable us to make the most effective data-driven decisions. 

This tutorial will show you how to utilize PySpark to connect Atlas with Databricks so you can take advantage of both platforms. 

MongoDB Atlas is a scalable and flexible storage solution for your data while Azure Databricks provides the power of Apache Spark to work with the security and collaboration features that are available with a Microsoft Azure subscription. [Apache Spark](https://spark.apache.org/) provides the Python interface for working with Spark, PySpark, which allows for an easy-to-use interface for developing in Python. To properly connect PySpark with MongoDB Atlas, the [MongoDB Spark Connector](https://www.mongodb.com/docs/spark-connector/current/) is utilized. This connector ensures for seamless compatibility, as you will see below in the tutorial. 

![Alt Text: technical diagram showing how Atlas, PySpark, the MongoDB Spark Connector, and Azure Databricks are integrated](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/databrickspic_6abce4187c.png)

Our tutorial to combine the above platforms will consist of viewing and manipulating an Atlas cluster and visualizing our data from the cluster back in our PySpark console. We will be setting up both Atlas and Azure Databricks clusters, connecting our Databricks cluster to our IDE, and writing scripts to view and contribute to the cluster in our Atlas account. Let’s get started!

### Requirements
In order to successfully recreate this project, please ensure you have everything in the following list: 

* [MongoDB Atlas](https://www.mongodb.com/atlas/database) account.

* [Microsoft Azure](https://azure.microsoft.com/en-us/products/databricks/) subscription (two-week free tier trial). 

* Python 3.8+.

* [GitHub Repository](https://github.com/mongodb-developer/atlas-pyspark-databricks).

* Java on your local machine.

## Setting up a MongoDB Atlas cluster
Our first step is to set up a MongoDB Atlas cluster. Access the Atlas UI and [follow these steps](https://www.mongodb.com/basics/clusters/mongodb-cluster-setup). For this tutorial, a free “shared” cluster is perfect. Create a database and name it “bookshelf” with a collection inside named “books”. To ensure ease for this tutorial, please allow for a connection from anywhere within your cluster’s network securities.

Once properly provisioned, your cluster will look like this:
![Alt Text: the initial set-up of our Atlas cluster](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/databricks1_f1bb39aaec.png)

Now we can set up our Azure Databricks cluster. 

## Setting up an Azure Databricks cluster

Access the [Azure Databricks](https://azure.microsoft.com/en-us/products/databricks/) page, sign in, and access the Azure Databricks tab. This is where you’ll create an Azure Databricks workspace. 

![Alt Text: the configuration of our Azure Databricks workspace](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2023_05_01_at_2_58_42_PM_f44cb7b2ab.png)

For our Databricks cluster, a free trial works perfectly for this tutorial. Once the cluster is provisioned, you’ll only have two weeks to access it before you need to upgrade. 

Hit “Review and Create” at the bottom. Once your workspace is validated, click “Create.” Once your deployment is complete, click on “Go to Resource.” You’ll be taken to your workspace overview. Click on “Launch Workspace” in the middle of the page. 

This will direct you to the Microsoft Azure Databricks UI where we can create the Databricks cluster. On the left-hand of the screen, click on “Create a Cluster,” and then click “Create Compute” to access the correct form. 

![Alt Text: the configuration of our Azure Databricks cluster](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2023_05_01_at_3_00_13_PM_dcf3d34ade.png)

When creating your cluster, pay close attention to what your “Databricks runtime version” is. Continue through the steps to create your cluster. 

We’re now going to install the libraries we need in order to connect to our MongoDB Atlas cluster. Head to the “Libraries” tab of your cluster, click on “Install New,” and select “Maven.” Hit “Search Packages” next to “Coordinates.” Search for `mongo` and select the `mongo-spark` package. Do the same thing with `xml` and select the `spark-xml` package. When done, your library tab will look like this: 
![Alt Text: libraries to install for your Databricks cluster](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2023_05_01_at_3_02_13_PM_5439367d65.png)

## Utilizing Databricks-Connect
Now that we have our Azure Databricks cluster ready, we need to properly connect it to our IDE. We can do this through a very handy configuration named Databricks Connect. Databricks Connect allows for Azure Databricks clusters to connect seamlessly to the IDE of your choosing.

### Databricks configuration essentials

Before we establish our connection, let’s make sure we have our configuration essentials. This is available in [the Databricks Connect tutorial](https://learn.microsoft.com/en-us/azure/databricks/dev-tools/databricks-connect) on Microsoft’s website under “Step 2: Configure connection properties.” Please note these properties down in a safe place, as you will not be able to connect properly without them.


### Databricks-Connect configuration
Access the Databricks Connect page linked above to properly set up `databricks-connect` on your machine. Ensure that you are downloading the `databricks-connect` version that is compatible with your Python version and is the same as the Databricks runtime version in your Azure cluster. 

>Please ensure prior to installation that you are working with a virtual environment for this project. Failure to use a virtual environment may cause PySpark package conflicts in your console. 

Virtual environment steps in Python:
```
python3 -m venv name 
```
Where the `name` is the name of your environment, so truly you can call it anything. 

Our second step is to activate our virtual environment:
```
source name/bin/activate 
```
And that’s it. We are now in our Python virtual environment. You can see that you’re in it when the little (name) or whatever you named it shows up.

* * *

Continuing on...for our project, use this installation command:
```
pip install -U “databricks-connect==10.4.*” 
```
Once fully downloaded, we need to set up our cluster configuration. Use the configure command and follow the instructions. This is where you will input your configuration essentials from our “Databricks configuration essentials” section.

Once finished, use this command to check if you’re connected to your cluster:
```
databricks-connect test
```

You’ll know you’re correctly configured when you see an “All tests passed” in your console. 
Now, it’s time to set up our SparkSessions and connect them to our Atlas cluster.

## SparkSession + Atlas configuration
The creation of a SparkSession object is crucial for our tutorial because it provides a way to access all important PySpark features in one place. These features include: reading data, creating data frames, and managing the overall configuration of PySpark applications. Our SparkSession will enable us to read and write to our Atlas cluster through the data frames we create.

The full code is on our [Github](https://github.com/mongodb-developer/atlas-pyspark-databricks) account, so please access it there if you would like to replicate this exact tutorial. We will only go over the code for some of the essentials of the tutorial below.

This is the SparkSession object we need to include. We are going to use a basic structure where we describe the application name, configure our “read” and “write” connectors to our `connection_string` (our MongoDB cluster [connection string](https://www.mongodb.com/docs/guides/atlas/connection-string/) that we have saved safely as an environment variable), and configure our `mongo-spark-connector`. Make sure to use the correct `mongo-spark-connector` for your environment. For ours, it is version 10.0.3. Depending on your Python version, the `mongo-spark-connector` version might be different. To find which version is compatible with your environment, please refer to the [MVN Repository documents](https://mvnrepository.com/artifact/org.mongodb.spark/mongo-spark-connector). 

```
# use environment variable for uri 
load_dotenv()
connection_string: str = os.environ.get("CONNECTION_STRING")

# Create a SparkSession. Ensure you have the mongo-spark-connector included.
my_spark = SparkSession \
    .builder \
    .appName("tutorial") \
    .config("spark.mongodb.read.connection.uri", connection_string) \
    .config("spark.mongodb.write.connection.uri", connection_string) \
    .config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector:10.0.3") \
    .getOrCreate()

```

For more help on how to create a SparkSession object with MongoDB and for more details on the `mongo-spark-connector`, please view the [documentation](https://www.mongodb.com/docs/spark-connector/current/getting-started/).

Our next step is to create two data frames, one to `write` a book to our Atlas cluster, and a second to `read` back all the books in our cluster. These data frames are essential; make sure to use the proper format or else they will not properly connect to your cluster. 

[Data frame to `write`](https://www.mongodb.com/docs/spark-connector/current/write-to-mongodb/) a book:
```
add_books = my_spark \
    .createDataFrame([("<title>", "<author>", <year>)], ["title", "author", "year"])

add_books.write \
    .format("com.mongodb.spark.sql.DefaultSource") \
    .option('uri', connection_string) \
    .option('database', 'bookshelf') \
    .option('collection', 'books') \
    .mode("append") \
    .save() 

```
[Data frame to `read`](https://www.mongodb.com/docs/spark-connector/current/read-from-mongodb/) back our books:
```
# Create a data frame so you can read in your books from your bookshelf.
return_books = my_spark.read.format("com.mongodb.spark.sql.DefaultSource") \
    .option('uri', connection_string) \
    .option('database', 'bookshelf') \
    .option('collection', 'books') \
    .load()

# Show the books in your PySpark shell.
return_books.show()

```

Add in the book of your choosing under the `add_books` dataframe. Here, exchange the title, author, and year for the areas with the `< >` brackets. Once you add in your book and run the file, you’ll see that the logs are telling us we’re connecting properly and we can see the added books in our PySpark shell. This demo script was run six separate times to add in six different books. A picture of the console is below:

![Alt Text: list of books from our Atlas cluster shown in our PySpark shell](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/databricksbooksconsole_3b5f6102fc.png)

We can double-check our cluster in Atlas to ensure they match up: 
![Alt Text: list of books in our Atlas cluster shown in our Atlas account](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/Screenshot_2023_05_01_at_2_46_38_PM_98f6cfe75b.png)

## Conclusion
Congratulations! We have successfully connected our MongoDB Atlas cluster to Azure Databricks through PySpark, and we can `read` and `write` data straight to our Atlas cluster. 

The skills you’ve learned from this tutorial will allow you to utilize Atlas’s scalable and flexible storage solution while leveraging Azure Databricks’ advanced analytics capabilities. This combination can allow developers to handle any amount of data in an efficient and scalable manner, while allowing them to gain insights into complex data sets to make exciting data-driven decisions! 

Questions? Comments? Let’s continue the conversation over at the [MongoDB Developer Community](https://www.mongodb.com/community/forums/)!

This tutorial will show you how to connect MongoDB Atlas to Azure Databricks using PySpark. 


 Utilizing PySpark to Connect MongoDB Atlas with Azure Databricks

Francesco Baldissera

Learn how to develop using an event-driven architecture that leverages MongoDB Atlas and Databricks.

How to Leverage an Event-Driven Architecture with MongoDB and Databricks

Robert Walters

MongoDB has released a version 10 of the MongoDB Connector for Apache Spark that leverages the new Spark Data Sources API V2 with support for Spark Structured Streaming.

## Why a new version?

The current version of the [MongoDB Spark Connector](https://www.mongodb.com/docs/spark-connector/current/) was originally written in 2016 and is based upon V1 of the Spark Data Sources API. While this API version is still supported, Databricks has released an updated version of the API, making it easier for data sources like MongoDB to work with Spark. By having the MongoDB Spark Connector use V2 of the API, an immediate benefit is a tighter integration with Spark [Structured Streaming](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#overview).

*Note: With respect to the previous version of the MongoDB Spark Connector that supported the V1 API, MongoDB will continue to support this release until such a time as Databricks depreciates V1 of the Data Source API. While no new features will be implemented, upgrades to the connector will include bug fixes and support for the current versions of Spark only.*

## What version should I use?

The new MongoDB Spark Connector release (Version 10.1) is not intended to be a direct replacement for your applications that use the previous version of MongoDB Spark Connector.

The new Connector uses a different namespace with a short name, “mongodb” (full path is  “com.mongodb.spark.sql.connector.MongoTableProvider”), versus “mongo” (full path of  “com.mongodb.spark.DefaultSource”). Having a different namespace makes it possible to use both versions of the connector within the same Spark application! This is helpful in unit testing your application with the new Connector and making the transition on your timeline.  

Also, we are changing how we version the MongoDB Spark Connector. The previous versions of the MongoDB Spark Connector aligned with the version of Spark that was supported—e.g., Version 2.4 of the MongoDB Spark Connector works with Spark 2.4. Keep in mind that going forward, this will not be the case. The MongoDB documentation will make this clear as to which versions of Spark the connector supports.

## Structured Streaming with MongoDB using continuous mode

Apache Spark comes with a stream processing engine called [Structured Streaming](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#overview), which is based on Spark's SQL engine and DataFrame APIs. Spark Structured Streaming treats each incoming stream of data as a micro-batch, continually appending each micro-batch to the target dataset. This makes it easy to convert existing Spark batch jobs into a streaming job. Structured Streaming has evolved over Spark releases and in Spark 2.3 introduced Continuous Processing mode, which took the micro-batch latency from over 100ms to about 1ms. Note this feature is still in experimental mode according to the [official Spark Documentation](https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#continuous-processing). In the following example, we’ll show you how to stream data between MongoDB and Spark using Structured Streams and continuous processing. First, we’ll look at reading data from MongoDB.

### Reading streaming data from MongoDB

You can stream data from MongoDB to Spark using the new Spark Connector. Consider the following example that streams stock data from a MongoDB Atlas cluster. A sample document in MongoDB is as follows:

```
{
  _id: ObjectId("624767546df0f7dd8783f300"),
  company_symbol: 'HSL',
  company_name: 'HUNGRY SYNDROME LLC',
  price: 45.74,
  tx_time: '2022-04-01T16:57:56Z'
}
```
In this code example, we will use the new MongoDB Spark Connector and read from the StockData collection. When the Spark Connector opens a streaming read connection to MongoDB, it opens the connection and creates a [MongoDB Change Stream](https://www.mongodb.com/docs/manual/changeStreams/) for the given database and collection. A change stream is used to subscribe to changes in MongoDB. As data is inserted, updated, and deleted, change stream events are created. It’s these [change events](https://www.mongodb.com/docs/manual/reference/change-events/) that are passed back to the client in this case the Spark application. There are configuration options that can change the structure of this event message. For example, if you want to return just the document itself and not include the change stream event metadata, set  “spark.mongodb.change.stream.publish.full.document.only” to true.

```
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.\
        builder.\
        appName("streamingExampleRead").\
        config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector_2.12::10.1.1').\
        getOrCreate()

query=(spark.readStream.format("mongodb")
.option('spark.mongodb.connection.uri', '<CONNECTION STRING>')
    	.option('spark.mongodb.database', 'Stocks') \
    	.option('spark.mongodb.collection', 'StockData') \
.option('spark.mongodb.change.stream.publish.full.document.only','true') \
    	.option("forceDeleteTempCheckpointLocation", "true") \
    	.load())

query.printSchema()
```

The schema is inferred from the MongoDB collection. You can see from the printSchema command that our document structure is as follows:


| root: |  |  |  |
| --- | --- | --- | --- |
|  |_id  |string  |(nullable=true)  |
|  |company_name| string  | (nullable=true) |
|  | company_symbol | string  | (nullable=true) |
|  | price | double | (nullable=true) |
|  | tx_time | string  | (nullable=true) |

We can verify that the dataset is streaming with the isStreaming command.

```
query.isStreaming
```

Next, let’s read the data on the console as it gets inserted into MongoDB.

```
query2=(query.writeStream \
  .outputMode("append") \
  .option("forceDeleteTempCheckpointLocation", "true") \
  .format("console") \
  .trigger(continuous="1 second")
  .start().awaitTermination());
```

When the above code was run through spark-submit, the output resembled the following:

… removed for brevity …

-------------------------------------------
Batch: 2
-------------------------------------------
+--------------------+--------------------+--------------+-----+-------------------+
|                 _id|        company_name|company_symbol|price|            tx_time|
+--------------------+--------------------+--------------+-----+-------------------+
|62476caa6df0f7dd8...| HUNGRY SYNDROME LLC|           HSL|45.99|2022-04-01 17:20:42|
|62476caa6df0f7dd8...|APPETIZING MARGIN...|           AMP|12.81|2022-04-01 17:20:42|
|62476caa6df0f7dd8...|EMBARRASSED COCKT...|           ECC|38.18|2022-04-01 17:20:42|
|62476caa6df0f7dd8...|PERFECT INJURY CO...|           PIC|86.85|2022-04-01 17:20:42|
|62476caa6df0f7dd8...|GIDDY INNOVATIONS...|           GMI|84.46|2022-04-01 17:20:42|
+--------------------+--------------------+--------------+-----+-------------------+

… removed for brevity …

-------------------------------------------

Batch: 3
-------------------------------------------
+--------------------+--------------------+--------------+-----+-------------------+
|                 _id|        company_name|company_symbol|price|            tx_time|
+--------------------+--------------------+--------------+-----+-------------------+
|62476cab6df0f7dd8...| HUNGRY SYNDROME LLC|           HSL|46.04|2022-04-01 17:20:43|
|62476cab6df0f7dd8...|APPETIZING MARGIN...|           AMP| 12.8|2022-04-01 17:20:43|
|62476cab6df0f7dd8...|EMBARRASSED COCKT...|           ECC| 38.2|2022-04-01 17:20:43|
|62476cab6df0f7dd8...|PERFECT INJURY CO...|           PIC|86.85|2022-04-01 17:20:43|
|62476cab6df0f7dd8...|GIDDY INNOVATIONS...|           GMI|84.46|2022-04-01 17:20:43|
+--------------------+--------------------+--------------+-----+-------------------+
 
### Writing streaming data to MongoDB

Next, let’s consider an example where we stream data from Apache Kafka to MongoDB. Here the source is a kafka topic “stockdata.Stocks.StockData.” As data arrives in this topic, it’s run through Spark with the message contents being parsed, transformed, and written into MongoDB. Here is the code listing with comments in-line:

```
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.functions import *
from pyspark.sql.types import StructType,TimestampType, DoubleType, StringType, StructField

spark = SparkSession.\
        builder.\
        appName("streamingExampleWrite").\
        config('spark.jars.packages', 'org.mongodb.spark:mongo-spark-connector:10.1.1').\
        config('spark.jars.packages', 'org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0').\
        getOrCreate()

df = spark \
  .readStream \
  .format("kafka") \
  .option("startingOffsets", "earliest") \
  .option("kafka.bootstrap.servers", "KAFKA BROKER HOST HERE") \
  .option("subscribe", "stockdata.Stocks.StockData") \
  .load()

schemaStock = StructType([ \
    StructField("_id",StringType(),True), \
    StructField("company_name",StringType(), True), \
    StructField("company_symbol",StringType(), True), \
    StructField("price",StringType(), True), \
    StructField("tx_time",StringType(), True)])

schemaKafka = StructType([ \
    StructField("payload",StringType(),True)])
```

Note that Kafka topic message arrives in this format -> key (binary), value (binary), topic (string), partition (int), offset (long), timestamp (long), timestamptype (int). See [Structured Streaming + Kafka Integration Guide (Kafka broker version 0.10.0 or higher)](https://spark.apache.org/docs/2.3.4/structured-streaming-kafka-integration.html) for more information on the Kafka and Spark integration.

To process the message for consumption into MongoDB, we want to pick out the value which is in binary format and convert it to JSON.

```
stockDF=df.selectExpr("CAST(value AS STRING)")
```

For reference, here is an example of an event (the value converted into a string) that is on the Kafka topic:

```
{
  "schema": {
    "type": "string",
    "optional": false
  },
  "payload": "{\"_id\": {\"$oid\": \"6249f8096df0f7dd8785d70a\"}, \"company_symbol\": \"GMI\", \"company_name\": \"GIDDY INNOVATIONS\", \"price\": 87.57, \"tx_time\": \"2022-04-03T15:39:53Z\"}"
}
```

We want to isolate the payload field and convert it to a JSON representation leveraging the shcemaStock defined above. For clarity, we have broken up the operation into multiple steps to explain the process. First, we want to convert the value into JSON.

```
stockDF=stockDF.select(from_json(col('value'),schemaKafka).alias("json_data")).selectExpr('json_data.*')
```

The dataset now contains data that resembles

```
…
 {
    _id: ObjectId("624c6206e152b632f88a8ee2"),
    payload: '{"_id": {"$oid": "6249f8046df0f7dd8785d6f1"}, "company_symbol": "GMI", "company_name": "GIDDY MONASTICISM INNOVATIONS", "price": 87.62, "tx_time": "2022-04-03T15:39:48Z"}'
  }, …
```

Next, we want to capture just the value of the payload field and convert that into JSON since it’s stored as a string.

```
stockDF=stockDF.select(from_json(col('payload'),schemaStock).alias("json_data2")).selectExpr('json_data2.*')
```

Now we can do whatever transforms we would like to do on the data. In this case, let’s convert the tx_time into a timestamp.

```
stockDF=stockDF.withColumn("tx_time",col("tx_time").cast("timestamp"))
```

The Dataset is in a format that’s ready for consumption into MongoDB, so let’s stream it out to MongoDB. To do this, use the writeStream method. Keep in mind there are various options to set. For example, when present, the “trigger” option processes the results in batches. In this example, it’s every 10 seconds. Removing the trigger field will result in continuous writing. For more information on options and parameters, check out the Structured Streaming Guide.

```
dsw = (
  stockDF.writeStream
    .format("mongodb")
    .queryName("ToMDB")
    .option("checkpointLocation", "/tmp/pyspark7/")
    .option("forceDeleteTempCheckpointLocation", "true")
    .option('spark.mongodb.connection.uri', ‘<CONNECTION STRING>')
    .option('spark.mongodb.database', 'Stocks')
    .option('spark.mongodb.collection', 'Sink')
    .trigger(continuous="10 seconds")
    .outputMode("append")
    .start().awaitTermination());
```

## Structured Streaming with MongoDB using Microbatch mode
While continuous mode offers a lot of promise in terms of the latency and performance characteristics, the support for various popular connectors like AWS S3 for example is non-existent.  Thus, you might end up using microbatch mode within your solution.  The key difference between the two is how spark handles obtaining the data from the stream.  As mentioned previously, the data is batched and processed versus using a continuous append to a table.  The noticeable difference is the advertised latency of microbatch around 100ms which for most workloads might not be an issue.
### Reading streaming data from MongoDB using microbatch

Unlike when we specify a write, when we read from MongoDB, there is no special configuration to tell Spark to use microbatch or continuous. This behavior is determined only when you write.  Thus, in our code example, to read from MongoDB is the same in both cases, e.g.:

```
query=(spark.readStream.format("mongodb").\
	option('spark.mongodb.connection.uri', '<<MONGODB CONNECTION STRING>>').\
	option('spark.mongodb.database', 'Stocks').\
	option('spark.mongodb.collection', 'StockData').\
	option('spark.mongodb.change.stream.publish.full.document.only','true').\
	option("forceDeleteTempCheckpointLocation", "true").\
	load())
```

Recall from the previous discussion on reading MongoDB data, when using `spark.readStream.format("mongodb")`, MongoDB opens a [change stream](https://www.mongodb.com/docs/manual/changeStreams/) and subscribes to changes as they occur in the database.  With microbatch each microbatch event opens a new change stream cursor making this form of microbatch streaming less efficient than continuous streams.  That said, some consumers of streaming data such as AWS S3 only support data from microbatch streams.

### Writing streaming data to MongoDB using microbatch
Consider the previous writeStream example code:

```
dsw = (
  stockDF.writeStream
    .format("mongodb")
    .queryName("ToMDB")
    .option("checkpointLocation", "/tmp/pyspark7/")
    .option("forceDeleteTempCheckpointLocation", "true")
    .option('spark.mongodb.connection.uri', '<<MONGODB CONNECTION STRING>>')
    .option('spark.mongodb.database', 'Stocks')
    .option('spark.mongodb.collection', 'Sink')
    .trigger(continuous="10 seconds")
    .outputMode("append")
    .start().awaitTermination());
```

Here the .trigger parameter was used to tell Spark to use Continuous mode streaming, to use microbatch simply remove the .trigger parameter.

## Go forth and stream!

Streaming data is a critical component of many types of applications. MongoDB has evolved over the years, continually adding features and functionality to support these types of workloads. With the MongoDB Spark Connector version 10.1, you can quickly stream data to and from MongoDB with a few lines of code.

For more information and examples on the new MongoDB Spark Connector version 10.1, check out the [online documentation](https://docs.mongodb.com/spark-connector). Have questions about the connector or MongoDB? Post a question in the [MongoDB Developer Community Connectors & Integrations](https://www.mongodb.com/community/forums/c/data/connectors-integrations/48) forum.

MongoDB has released a new spark connector, MongoDB Spark Connector V10. In this article, learn how to read from and write to MongoDB through Spark Structured Streaming.

Streaming Data with Apache Spark and MongoDB

Rachelle Palmer

## Creators
[Lucas De Oliveira](https://www.linkedin.com/in/lbdeoliveira/), [Chandrish Ambati](https://www.linkedin.com/in/avmchandrish/), and [Anish Mukherjee](https://www.linkedin.com/in/mukherjeeanish/) from University of San Francisco contributed this amazing project.

## Background to the Project
In 2018, Spotify organized an Association for Computing Machinery (ACM) [RecSys Challenge](https://www.recsyschallenge.com/2018/) where they posted [a dataset of one million playlists](https://www.aicrowd.com/challenges/spotify-million-playlist-dataset-challenge), challenging participants to recommend a list of 500 songs given a user-created playlist.

As both music lovers and data scientists, we were naturally drawn to this challenge. Right away, we agreed that combining song embeddings with some nearest-neighbors method for recommendation would likely produce very good results. Importantly, we were curious about how we could  solve  this recommendation task at scale with over [4 billion user-curated playlists](https://soundplate.com/how-many-playlists-are-there-on-spotify-and-other-spotify-stats/) on Spotify, where this number keeps growing. This realization raised serious questions about how to train a decent model since all that data would likely not fit in memory or a single server.

## What We Built
This project resulted in a scalable ETL pipeline utilizing
* Apache Spark
* MongoDB
* Amazon S3
* Databricks (PySpark)

These were used to train a deep learning Word2Vec model to build song and playlist embeddings for recommendation. We followed up with data visualizations we created on Tensorflow’s Embedding Projector.

## The Process
### Collecting Lyrics
The most tedious task of this project was collecting as many lyrics for the songs in the playlists as possible. We began by isolating the unique songs in the playlist files by their track URI; in total we had over 2 million unique songs. Then, we used the track name and artist name to look up the lyrics on the web. Initially, we used simple Python requests to pull in the lyrical information but this proved too slow for our purposes. We then used asyncio, which allowed us to make requests concurrently. This sped up the process significantly, reducing the downloading time of lyrics for 10k songs from 15 mins to under a minute. Ultimately, we were only able to collect lyrics for 138,000 songs.

### Pre-processing
The original dataset contains 1 million playlists spread across 1 thousand JSON files totaling about 33 GB of data. We used PySpark in Databricks to preprocess these separate JSON files into a single SparkSQL DataFrame and then joined this DataFrame with the lyrics we saved. 

While the aforementioned data collection and preprocessing steps are time-consuming, the model also needs to be re-trained and re-evaluated often, so it is critical to store data in a scalable database. In addition, we’d like to consider a database that is schemaless for future expansion in data sets and supports various data types.  Considering our needs, we concluded that MongoDB would be the optimal solution as a data and feature store.

Check out the [Preprocessing.ipynb](https://github.com/lbdeoliveira/song-playlist-recommendation/blob/main/notebooks/Preprocessing_Pipeline.ipynb) notebook to see how we preprocessed the data.

### Training Song Embeddings
For our analyses, we read our preprocessed data from MongoDB into a Spark DataFrame and grouped the records by playlist id (pid), aggregating all of the songs in a playlist into a list under the column song_list. 
Using the Word2Vec model in Spark MLlib we trained song embeddings by feeding lists of track IDs from a playlist into the model much like you would send a list of words from a sentence to train word embeddings. As shown below, we trained song embeddings in only 3 lines of PySpark code:
```
from pyspark.ml.feature import Word2Vec
word2Vec = Word2Vec(vectorSize=32, seed=42, inputCol="song_list").setMinCount(1)
word2Vec.sexMaxIter(10)
model = word2Vec.fit(df_play)
```

We then saved the song embeddings down to MongoDB for later use. Below is a snapshot of the song embeddings DataFrame that we saved:

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

Check out the [Song_Embeddings.ipynb](https://github.com/lbdeoliveira/song-playlist-recommendation/blob/main/notebooks/Song_Embeddings.ipynb) notebook to see how we train song embeddings.

### Training Playlists Embeddings
Finally, we extended our recommendation task beyond simple song recommendations to recommending entire playlists. Given an input playlist, we would return the k closest or most similar playlists. We took a “continuous bag of songs” approach to this problem by calculating playlist embeddings as the average of all song embeddings in that playlist.

This workflow started by reading back the song embeddings from MongoDB into a SparkSQL DataFrame. Then, we calculated a playlist embedding by taking the average of all song embeddings in that playlist and saved them in MongoDB.

Check out the [Playlist_Embeddings.ipynb](https://github.com/lbdeoliveira/song-playlist-recommendation/blob/main/notebooks/Playlist_Embeddings.ipynb) notebook to see how we did this.

### Training Lyrics Embeddings
Are you still reading? Whew!

We trained lyrics embeddings by loading in a song's lyrics, separating the words into lists, and feeding those words to a Word2Vec model to produce 32-dimensional vectors for each word. We then took the average embedding across all words as that song's lyrical embedding. Ultimately, our analytical goal here was to determine whether users create playlists based on common lyrical themes by seeing if the pairwise song embedding distance and the pairwise lyrical embedding distance between two songs were correlated. Unsurprisingly, it appears they are not.

Check out the [Lyrical_Embeddings.ipynb](https://github.com/lbdeoliveira/song-playlist-recommendation/blob/main/notebooks/Lyrical_Embeddings.ipynb) notebook to see our analysis.

## Notes on our Approach
You may be wondering why we used a language model (Word2Vec) to train these embeddings. Why not use a Pin2Vec or custom neural network model to predict implicit ratings? For practical reasons, we wanted to work exclusively in the Spark ecosystem and deal with the data in a distributed fashion. This was a constraint set on the project ahead of time and challenged us to think creatively.

However, we found Word2Vec an attractive candidate model for theoretical reasons as well. The Word2Vec model uses a word’s context to train static embeddings by training the input word’s embeddings to predict its surrounding words. In essence, the embedding of any word is determined by how it co-occurs with other words. This had a clear mapping to our own problem: by using a Word2Vec model the distance between song embeddings would reflect the songs’ co-occurrence throughout 1M playlists, making it a useful measure for a distance-based recommendation (nearest neighbors). It would effectively model how people grouped songs together, using user behavior as the determinant factor in similarity.

Additionally, the Word2Vec model accepts input in the form of a list of words. For each playlist we had a list of track IDs, which made working with the Word2Vec model not only conceptually but also practically appealing.

## Data Visualizations with Tensorflow and MongoDB
After all of that, we were finally ready to visualize our results and make some interactive recommendations. We decided to represent our embedding results visually using [Tensorflow’s Embedding Projector](https://projector.tensorflow.org) which maps the 32-dimensional song and playlist embeddings into an interactive visualization of a 3D embedding space. You have the choice of using PCA or tSNE for dimensionality reduction and cosine similarity or Euclidean distance for measuring distances between vectors.

Click [here](https://projector.tensorflow.org/?config=https://embedding-projector.s3.us-west-2.amazonaws.com/template_config_2M.json) for the song embeddings projector for the full 2 million songs, or [here](https://projector.tensorflow.org/?config=https://embedding-projector.s3.us-west-2.amazonaws.com/template_config.json) for a less crowded version with a random sample of 100k songs (shown below):

![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/164569070_88be5d94_a291_495d_b33d_3094634d9205_afc69037e9.gif)

The neat thing about using Tensorflow’s projector is that it gives us a beautiful visualization tool and distance calculator all in one. Try searching on the right panel for a song and if the song is part of the original dataset, you will see the “most similar” songs appear under it.

## Using MongoDB for ML/AI
We were impressed by how easy it was to use MongoDB to reliably store and load our data. Because we were using distributed computing, it would have been infeasible to run our pipeline from start to finish any time we wanted to update our code or fine-tune the model. MongoDB allowed us to save our incremental results for later processing and modeling, which collectively saved us hours of waiting for code to re-run.

It worked well with all the tools we use everyday and the tooling we chose - we didn't have any areas of friction. 

We were shocked by how this method of training embeddings actually worked. While the 2 million song embedding projector is crowded visually, we see that the recommendations it produces are actually quite good at grouping songs together.

Consider the embedding recommendation for The Beatles’ “A Day In The Life”:
![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/165581471_420ad43d_47f9_4aee_bf0c_2de6ac301144_c656a07f0e.gif)

Or the recommendation for Jay Z’s “Heart of the City (Ain’t No Love)”:
![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/164567916_28f6ebd6_0473_4d7b_b60c_55af628016f7_ec16751581.png)

Fan of Taylor Swift? Here are the recommendations for “New Romantics”:
![](https://mongodb-devhub-cms.s3.us-west-1.amazonaws.com/164567889_9ccf8326_7bc8_4c8e_9bc6_f123c0acba3f_0fab2ea1db.png)

We were delighted to find naturally occurring clusters in the playlist embeddings. Most notably, we see a cluster containing mostly Christian rock, one with Christmas music, one for reggaeton, and one large cluster where genres span its length rather continuously and intuitively.

Note also that when we select a playlist, we have many recommended playlists with the same names. This in essence validates our song embeddings. Recall that playlist embeddings were created by taking the average embedding of all its songs; the name of the playlists did not factor in at all. The similar names only conceptually reinforce this fact.



## Next Steps?
We felt happy with the conclusion of this project but there is more that could be done here.

1. We could use these trained song embeddings in other downstream tasks and see how effective these are. Also, you could download the song embeddings we here: [Embeddings](https://embedding-projector.s3.us-west-2.amazonaws.com/emb_word2vec_2M.tsv) | [Meta Info](https://embedding-projector.s3.us-west-2.amazonaws.com/meta_word2vec_2M.tsv)
2. We could look at other methods of training these embeddings using some recurrent neural networks and enhanced implementation of this Word2Vec model.





Python code example application for Spotify playlist and song recommendations using spark and tensorflow

Spark

All Spark Content

How to Leverage an Event-Driven Architecture with MongoDB and Databricks

Utilizing PySpark to Connect MongoDB Atlas with Azure Databricks

A Spotify Song and Playlist Recommendation Engine

Streaming Data with Apache Spark and MongoDB

MongoDB Podcast Interview with Connectors and Translators Team

Related Topics