samples/notebooks/Spark+and+Cosmos+DB+Change+Feed.ipynb

{"nbformat_minor": 2, "cells": [{"source": "# Connecting Spark with Cosmos DB Change feed  \nIn this sample, we connect to Cosmos DB change feed through the Cosmos DB Java SDK and Spark. We will read new changes and count the number of new changes. In this specific example, the collection we are reading from contains documents with product information. To learn more about Azure Cosmos DB and change feed check out the [main page](https://docs.microsoft.com/azure/cosmos-db/introduction), or the [change feed page](https://docs.microsoft.com/azure/cosmos-db/change-feed). ", "cell_type": "markdown", "metadata": {}}, {"source": "## Configuration\nConfiguring environment with Cosmos DB jar files available here on the [Github](https://github.com/Azure/azure-cosmosdb-spark/tree/master/releases/azure-cosmosdb-spark_2.1.0_2.11-0.0.4) ", "cell_type": "markdown", "metadata": {}}, {"execution_count": 1, "cell_type": "code", "source": "%%configure\n{ \"name\":\"Spark-to-Cosmos_DB_Connector\", \n  \"executorMemory\": \"8G\", \n  \"executorCores\": 2, \n  \"numExecutors\": 2, \n  \"driverCores\": 2,\n  \"jars\": [\"wasb:https:///example/jars/0.0.3.2/azure-documentdb-1.12.0.jar\",\"wasb:https:///example/jars/0.0.3.2/azure-cosmosdb-spark-0.0.3-SNAPSHOT.jar\", \"wasb:https:///example/jars/0.0.3.2/azure-documentdb-rx-0.9.0-rc1.jar\", \"wasb:https:///example/jars/0.0.3.2/rxjava-1.3.0.jar\" ],\n  \"conf\": {\n    \"spark.jars.packages\": \"graphframes:graphframes:0.5.0-spark2.1-s_2.11\",   \n    \"spark.jars.excludes\": \"org.scala-lang:scala-reflect\"\n   }\n}", "outputs": [{"output_type": "display_data", "data": {"text/plain": "<IPython.core.display.HTML object>", "text/html": "Current session configs: <tt>{u'kind': 'pyspark', u'name': u'Spark-to-Cosmos_DB_Connector', u'numExecutors': 2, u'conf': {u'spark.jars.packages': u'graphframes:graphframes:0.5.0-spark2.1-s_2.11', u'spark.jars.excludes': u'org.scala-lang:scala-reflect'}, u'executorCores': 2, u'driverCores': 2, u'jars': [u'wasb:https:///example/jars/0.0.3.2/azure-documentdb-1.12.0.jar', u'wasb:https:///example/jars/0.0.3.2/azure-cosmosdb-spark-0.0.3-SNAPSHOT.jar', u'wasb:https:///example/jars/0.0.3.2/azure-documentdb-rx-0.9.0-rc1.jar', u'wasb:https:///example/jars/0.0.3.2/rxjava-1.3.0.jar'], u'executorMemory': u'8G'}</tt><br>"}, "metadata": {}}, {"output_type": "display_data", "data": {"text/plain": "<IPython.core.display.HTML object>", "text/html": "<table>\n<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>130</td><td>application_1503579980138_0005</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"http:https://hn0-sparki.qz2e5wj10qmu3df2lfbmfh3ggb.gx.internal.cloudapp.net:8088/proxy/application_1503579980138_0005/\">Link</a></td><td><a target=\"_blank\" href=\"http:https://10.0.0.4:30060/node/containerlogs/container_e18_1503579980138_0005_01_000001/livy\">Link</a></td><td></td></tr></table>"}, "metadata": {}}], "metadata": {"collapsed": false}}, {"source": "## Connecting to Cosmos DB change feed \nTo start the Spark session with change feed, we need to specify some config options. \n\n**Endpoint**: your Cosmos DB url (i.e. https://youraccount.documents.azzure.com:443/)\n\n**Masterkey**: the key string for you Cosmos DB account\n\n**Database**: name of your existing database\n\n**Collection**: name of your existing collection from which you wish to read the change feed\n\n**ChangeFeedQueryName**: unique string for your query \n", "cell_type": "markdown", "metadata": {"collapsed": true}}, {"execution_count": 2, "cell_type": "code", "source": "# Adding variables \nrollingChangeFeed = False\nstartFromTheBeginning = False\nuseNextToken = True \n\nproductsConfig = {\n\"Endpoint\" : \"https://youraccount.documents.azure.com:443/\",\n\"Masterkey\" : \"mXBmwss4FJdsfhsdkjfhkeJFEDSQBNMFNUOGQsGCEeLRRkAZEUWebg==\",\n\"Database\" : \"AW\",\n\"Collection\" : \"AWProducts\", \n\"ReadChangeFeed\" : \"true\",\n\"ChangeFeedQueryName\" : str(rollingChangeFeed) + str(startFromTheBeginning) + str(useNextToken),\n\"ChangeFeedStartFromTheBeginning\" : str(startFromTheBeginning),\n\"ChangeFeedUseNextToken\" : str(useNextToken),\n\"RollingChangeFeed\" : str(rollingChangeFeed),\n\"ChangeFeedCheckpointLocation\" : \"./changefeedcheckpointlocation\",\n\"SamplingRatio\" : \"1.0\"\n}", "outputs": [{"output_type": "stream", "name": "stdout", "text": "Starting Spark application\n"}, {"output_type": "display_data", "data": {"text/plain": "<IPython.core.display.HTML object>", "text/html": "<table>\n<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>142</td><td>application_1503579980138_0017</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"http:https://hn0-sparki.qz2e5wj10qmu3df2lfbmfh3ggb.gx.internal.cloudapp.net:8088/proxy/application_1503579980138_0017/\">Link</a></td><td><a target=\"_blank\" href=\"http:https://10.0.0.20:30060/node/containerlogs/container_e18_1503579980138_0017_01_000001/livy\">Link</a></td><td>\u2714</td></tr></table>"}, "metadata": {}}, {"output_type": "stream", "name": "stdout", "text": "SparkSession available as 'spark'.\n"}], "metadata": {"collapsed": false}}, {"source": "Loading data frame ", "cell_type": "markdown", "metadata": {}}, {"execution_count": 3, "cell_type": "code", "source": "products = spark.read.format(\"com.microsoft.azure.cosmosdb.spark\").options(**productsConfig).load()", "outputs": [], "metadata": {"collapsed": false}}, {"source": "Initializing count of new products on change feed to be 0 ", "cell_type": "markdown", "metadata": {}}, {"execution_count": 4, "cell_type": "code", "source": "new_product_count = 0 ", "outputs": [], "metadata": {"collapsed": false}}, {"source": "Rerun below section to check for new changes and increment count of new changes. ", "cell_type": "markdown", "metadata": {}}, {"execution_count": 8, "cell_type": "code", "source": "products.show()\nnew_product_count += products.count() ", "outputs": [{"output_type": "stream", "name": "stdout", "text": "+-------------+--------------------+--------+--------------------+-----+--------------------+------+----+-------------+------------+-----------+---+--------------------+----------+-------+-------+----------+------------+-------------+--------------------+-----+\n|ProductNumber|               _etag|Category|                _rid|Model|         Description|Rating|Size|SubCategories|_attachments|SellEndDate| id|               _self|CategoryId|DocType| Weight|       _ts|CategoryName|SellStartDate|                Name|Color|\n+-------------+--------------------+--------+--------------------+-----+--------------------+------+----+-------------+------------+-----------+---+--------------------+----------+-------+-------+----------+------------+-------------+--------------------+-----+\n|   FR-R92B-58|\"06000400-0000-00...|    null|01NUAOs2KABDAQAAA...| null|Our lightest and ...|  null|  58|         null|attachments/|       null|700|dbs/01NUAA==/coll...|      null|   null|1016.04|1503685803|        null|         null|HL Road Frame - B...|Black|\n+-------------+--------------------+--------+--------------------+-----+--------------------+------+----+-------------+------------+-----------+---+--------------------+----------+-------+-------+----------+------------+-------------+--------------------+-----+"}], "metadata": {"scrolled": true, "collapsed": false}}, {"execution_count": 9, "cell_type": "code", "source": "print new_product_count", "outputs": [{"output_type": "stream", "name": "stdout", "text": "1"}], "metadata": {"collapsed": false}}], "nbformat": 4, "metadata": {"kernelspec": {"display_name": "PySpark", "name": "pysparkkernel", "language": ""}, "language_info": {"mimetype": "text/x-python", "pygments_lexer": "python2", "name": "pyspark", "codemirror_mode": {"version": 2, "name": "python"}}}}