<jupyter_start><jupyter_text>Getting started with Azure Cosmos DB's API for MongoDB and Synapse LinkIn this sample we will execute the following tasks:1. Insert a dataset using the traditional MongoDB client.1. Execute aggregation queries against the Analytical Store from the transactional data we inserted.1. Insert another dataset, but this time using a different datatype for the timestamp property.1. Execute aggregation queries again, consolidating both datasets. Pre-requisites1. Have you created a MongoDB API account in Azure Cosmos DB? If not, go to [Create an account for Azure Cosmos DB's API for MongoDB]().1. For your Cosmos DB account, have you enabled Synapse Link? If not, go to [Enable Synapse Link for Azure Cosmos DB's API for MongoDB]().1. Have you created a Synapse Workspace? If not, go to [Create Synapse Workspace account](). Please don't forget to add yourself as **Storage Blob Data Contributor** to the primary ADLS G2 account that is linked to the Synapse workspace. Create a Cosmos DB collection with analytical store enabledPlease be careful, all commands are case sensitive.1. Create a database named `DemoSynapseLinkMongoDB`. 1. Create a collection named `HTAP` with a Shard key called `item`. Make sure you set the `Analytical store` option to `On` when you create your collection. Optional - Connect your collection to SynapseTo accelerate future work, you can connect your collection to Synapse. **We won't use this capability in this demo**, but fell free to test and use it.1. Go to your Synapse Analytics workspace.1. Create a `Linked Data` connection for your MongoDB API account.     1. Under the `Data` blade, select the + (plus) sign.    1. Select the `Connect to external data` option.    1. Now select the `Azure Cosmos DB (MongoDB API)` option.     1. Enter all the information regarding your specific Azure Cosmos DB account either by using the dropdowns or by entering the connection string. Take note of the name you assigned to your `Linked Data` connection.     - Alternatively, you can also use the connection parameters from your account overview.1. Test the connection by looking for your database accounts in the `Data` blade, and under the `Linked` tab.    - There should be a list that contains all accounts and collections.    - Collections that have an `Analytical Store` enabled will have a distinctive icon.  Let's get the environment readyThis environment allows you to install and use any python libraries that you want to run. For this sample, you need to add the following libraries to your Spark pool:```pymongo==2.8.1aenum==2.2.4bson==0.5.10```Learn how to import libraries into your Spark pools in [this article](https://docs.microsoft.com/en-us/azure/synapse-analytics/spark/apache-spark-azure-portal-add-libraries). Please use the `requirements.txt` file located in the same folder of this notebook to update your pool packages.You can execute the following command to make sure all the libraries are installed correctly:<jupyter_code>import importlib

packages = ['pymongo','bson','aenum']
for package in packages:
    test = importlib.util.find_spec(package)
    if test:
        print(package, "OK")
    else:
        print(package, "PROBLEM - NOK")<jupyter_output><empty_output><jupyter_text>Add your database account and collection details here!<jupyter_code>DATABASE_ACCOUNT_NAME = 'your-cosmos-db-mongodb-account-name'
DATABASE_ACCOUNT_READWRITE_KEY = 'your-cosmos-db-mongodb-account-key'
DATABASE_NAME = 'DemoSynapseLinkMongoDB'
COLLECTION_NAME = 'HTAP'<jupyter_output><empty_output><jupyter_text>Let's initialize the MongoDB clientYou are only going to need the following parameters from your account overview: - Connection string.- Primary or secondary ready/write key.Remember that we named our database `DemoSynapseLinkMongoDB` and our collection `HTAP`.The code snippet below shows how to initialize the `MongoClient` object.<jupyter_code>from pymongo import MongoClient
from bson import ObjectId # For ObjectId to work

client = MongoClient("mongodb://{account}.mongo.cosmos.azure.com:10255/?ssl=true&replicaSet=globaldb".format(account = DATABASE_ACCOUNT_NAME)) # Your own database account endpoint.
db = client.DemoSynapseLinkMongoDB    # Select the database
db.authenticate(name=DATABASE_ACCOUNT_NAME,password=DATABASE_ACCOUNT_READWRITE_KEY) # Use your database account name and any of your read/write keys.<jupyter_output><empty_output><jupyter_text>Inserting data with the MongoClient driverThe following sample will generate 500 items based on random data. Each item will contain the following fields:- item, string- price, float- rating, integer- timestamp, [epoch integer](http://unixtimestamp.50x.eu/about.php)This data will be inserted into the MongoDB store of your database. This emulates the transactional data that an application would generate.<jupyter_code>from random import randint
import time

orders = db["HTAP"]

items = ['Pizza','Sandwich','Soup', 'Salad', 'Tacos']
prices = [2.99, 3.49, 5.49, 12.99, 54.49]

for x in range(1, 501):
    order = {
        'item' : items[randint(0, (len(items)-1))],
        'price' : prices[randint(0, (len(prices)-1))],
        'rating' : randint(1, 5),
        'timestamp' : time.time()
    }
    
    result=orders.insert(order)

print('finished creating 500 orders')<jupyter_output><empty_output><jupyter_text>Read data from the Analytical Store.Now that we have inserted some transactional data, let's read it from Azure Cosmos DB analytical store. Cosmos DB will automatically transform the BSON data (Binary JSON) into a columnar format, which will make it fast and easy to execute aggregation workloads on top of your transactional data, at no RUs or performance costs.The cells below will:1. Load the data from analytical store into a DataFrame.1. Check the top rows. Yes, the BSON data was converted into columar structured format.1. Check the DataFrame schema.1. Run aggregations> If you get an "no snapshot" error, Please check if your container was created with **analytical store** enabled. If your DataFrame has no data, please wait a couple of minutes because the root cause is that the auto sync between transactional and analytical stores isn't completed yet. This process usually takes 2 minutes, but in some cases it may take up to 5 minutes. Please wait a few minutes and run the command below again.**Important: Please note that we are using random values for prices and ratings. Don't expect the same results of the outputs below. What you can expect is the same behavior and experience.**<jupyter_code># Load the data from analytical store into a DataFrame.
df = spark.read.format("cosmos.olap")\
    .option("spark.cosmos.accountEndpoint", "https://{account}.documents.azure.com:443/".format(account = DATABASE_ACCOUNT_NAME))\
    .option("spark.cosmos.accountKey", DATABASE_ACCOUNT_READWRITE_KEY)\
    .option("spark.cosmos.database", DATABASE_NAME)\
    .option("spark.cosmos.container", COLLECTION_NAME)\
    .load()

# Checking the data
display(df)<jupyter_output><empty_output><jupyter_text>First Schema Analysis

Let's run the command below and check the schema of the `df` DataFrame that we just created and loaded. Please note that all properties of our document (item, price, rating, and timestamp) are represented in the DataFrame as a `struct` with one datatype within each one of them. This will change in the next cells, and to understand that is part of the learning objectives of this notebook.<jupyter_code>df.printSchema()<jupyter_output><empty_output><jupyter_text>Agregations

Now let's run aggregations on top of the `df` DataFrame that we just created.<jupyter_code># Run aggregations

df.groupBy(df.item.string).sum().show()<jupyter_output><empty_output><jupyter_text>Important - Spark Syntax for Aggregations on DataFrames

For the aggregation above, a syntax that doesn't explicity mentions the datatypes, like `df.groupBy(df['item']).sum().show()`, executes without an error. But it is **not recommended!** 

It runs because Spark automatically flattens the structure into an Array, where it takes each distinct value in the `struct` dict and applies the aggregation function. But you will see in the next cells that we may have more than one datatype for the same struct of a property, and the implicit conversion that Spark does can cause wrong results.
 Schema Representation - A quick note about the MongoDB schema in analytical store

Please note in the result above that for the `timestamp` field we have only 1 datatype: `struct`. 
We will see that this detail will change since we will insert data with different datatype for that `timestamp` field.

For Azure Cosmos DB API for MongoDB accounts, we make use of a **Full Fidelity Schema** as a default option. This is a representation of property names extended with their data types to provide an accurate 
representation of their values and avoid ambiguity.

This is why, when we called the fields above, we used their datatype as a suffix. Like in the example below:

```
df.filter((df.item.string == "Pizza")).show(10)
```

Notice how we specified the `string` type after the name of the property. Here is a map of all potential properties and their suffix representations in the Analytical Store:

| Original Data Type     &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;| Suffix    &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;| Example &nbsp;&nbsp;&nbsp;&nbsp; | 
|---------------|----------------|--------|
| Double        | ".float64"     |  `24.99`   |
| Array         | ".array"       |  `["a", "b"]`   |
| Binary        | ".binary"      |  `0`   |
| Boolean       | ".bool"        |  `True`   |
| Int32         | ".int32"       |  `123`   |
| Int64         | ".int64"       |  `255486129307`   |
| Null          | ".null"        |  `null`   |
| String        | ".string"      |  `"ABC"`   |
| Timestamp     | ".timestamp"   |  `Timestamp(0, 0)`   |
| DateTime      | ".date"        |  `ISODate("2020-08-21T07:43:07.375Z")`   |
| ObjectId      | ".objectId"    |  `ObjectId("5f3f7b59330ec25c132623a2")`   |
| Document      | ".object"      |  `{"a": "a"}`   |

These types are inferred from the data that is inserted in the transactional store. You can see the schema by executing the following command:
```
df.printSchema
```

> The default option for Azure Cosmos DB CORE (SQL) API, **Well defined Schema** is the default option. For more information about schemas representation, click [here](https://docs.microsoft.com/azure/cosmos-db/analytical-store-introductionschema-representation) .

  Let's insert more orders!This time we will use slightly different data. Each item will contain the following fields:- item, string- price, float- rating, integer- timestamp, [ISO String format](https://en.wikipedia.org/wiki/ISO_8601)Notice how the `Timestamp` field is now in a string format. This will help us understand how the different data fields can be read based on their data type.After that, we will load the data, check the schema, and run some queries.<jupyter_code>from random import randint
from time import strftime

orders = db["HTAP"]

items = ['Pizza','Sandwich','Soup', 'Salad', 'Tacos']
prices = [2.99, 3.49, 5.49, 12.99, 54.49]

for x in range(1, 501):
    order = {
        'item' : items[randint(0, (len(items)-1))],
        'price' : prices[randint(0, (len(prices)-1))],
        'rating' : randint(1, 5),
        'timestamp' : strftime("%Y-%m-%d %H:%M:%S")
    }
    
    result=orders.insert(order)

print('finished creating 500 orders')<jupyter_output><empty_output><jupyter_text>Let's reload the DataFrame and check the schema again!<jupyter_code># Load the Analytical Store data into a dataframe
# Make sure to run the cell with the secrets to get the DATABASE_ACCOUNT_NAME and the DATABASE_ACCOUNT_READWRITE_KEY variables.
df = spark.read.format("cosmos.olap")\
    .option("spark.cosmos.accountEndpoint", "https://{account}.documents.azure.com:443/".format(account = DATABASE_ACCOUNT_NAME))\
    .option("spark.cosmos.accountKey", DATABASE_ACCOUNT_READWRITE_KEY)\
    .option("spark.cosmos.database", DATABASE_NAME)\
    .option("spark.cosmos.container", COLLECTION_NAME)\
    .load()

# Check the schema AGAIN. Try to find something different.
df.printSchema()<jupyter_output><empty_output><jupyter_text>Schema Representation - What Changed?

Please note in the result above that now, for the `timestamp` field, we have 2 datatypes: `struct` and `string:string`. That happened because we added data with a different datatype. That's `Full Fidelity Schema`, when Azure Cosmos DB will do a full representation of your data, with the datatypes you used.

> If the result doesn't show two datatypes for `timestamp`, then wait a few minutes because the backend auto-sync process has not yet occurred.

 Queries

Now let's run some interesting queries, using the datypes to filter the data.<jupyter_code># SQL!!
# Let's see the data for pizzas that have a string timestamp
df.createOrReplaceTempView("Pizza")
sql_results = spark.sql("SELECT sum(price.float64),count(*) FROM Pizza where timestamp.string is not null and item.string = 'Pizza'")
sql_results.show()
# SQL!!
# Let's see the data for pizzas that have a string timestamp
df.createOrReplaceTempView("Pizza")
sql_results = spark.sql("SELECT sum(price.float64),count(*) FROM Pizza where timestamp.float64 is not null and item.string = 'Pizza'")
sql_results.show()
# SQL!!
# Let's compare both timestamp columns
df.createOrReplaceTempView("Pizza")
sql_results = spark.sql("SELECT max(timestamp.float64),max(timestamp.string) FROM Pizza where item.string = 'Pizza'")
sql_results.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09401272022808685, 'fraction_numerical': 0.013963008991885371, 'mean_word_length': 4.439363817097416, 'pattern_counts': {'":': 1, '<': 41, '<?xml version=': 0, '>': 44, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '430673', 'license': 'MIT', 'n_tokens_mistral': 4004, 'n_tokens_neox': 3797, 'n_words': 1603, 'path': 'Notebooks/PySpark/Synapse Link for Cosmos DB samples/MongoDB/spark-notebooks/pyspark/01-CosmosDBSynapseMongoDB.ipynb', 'repo_name': 'AFFogarty/Synapse', 'size': '13679'}