notebooks/elt-blob-storage-cosmosdb-python.ipynb

{"cells":[{"cell_type":"markdown","source":["# ELT Sample: Azure Blob Stroage - Databricks - CosmosDB\nIn this notebook, you extract data from Azure Blob Storage into Databricks cluster, run transformations on the data in Databricks cluster, and then load the transformed data into Azure Cosmos DB.\n## prerequisites:\n- Azure Blob Storage Account and Containers\n- Databricks Cluster (Spark)\n- Cosmos DB Spark Connector (azure-cosmosdb-spark)\n  - Create a library using maven coordinates. Simply typed in `azure-cosmosdb-spark_2.2.0` in the search box and search it, or create library by simply uploading jar file that can be donwload from marven central repository\n- Azure Cosmos DB Collection\n## Sample data\n- https://github.com/Azure/usql/blob/master/Examples/Samples/Data/json/radiowebsite/small_radio_json.json\n## LINKS\n- https://docs.azuredatabricks.net/spark/latest/data-sources/azure/azure-storage.html\n- https://github.com/Azure/azure-cosmosdb-spark"],"metadata":{}},{"cell_type":"markdown","source":["# Connecting to Azure Blob Storage and access a sample Json file"],"metadata":{}},{"cell_type":"markdown","source":["## Set up an account access key"],"metadata":{}},{"cell_type":"code","source":["# spark.conf.set(\n#  \"fs.azure.account.key.<storage-account-name>.blob.core.windows.net\",\n#  \"<storage-access-key>\")\n\nspark.conf.set(\n  \"fs.azure.account.key.databrickstore.blob.core.windows.net\",\n  \"S1PtMWvUw5If1Z8FMzXAxC7OMw9G5Go8BGCXJ81qpFVYpZ9dpXOnU4zlg0PbldKkbLIbmbv02WoJsgYLGKIfgg==\")"],"metadata":{},"outputs":[],"execution_count":4},{"cell_type":"markdown","source":["Once an account access key or a SAS is set up in your notebook, you can use standard Spark and Databricks APIs to read from the storage account"],"metadata":{}},{"cell_type":"code","source":["#dbutils.fs.ls(\"wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>\")\ndbutils.fs.ls(\"wasbs://dbdemo01@databrickstore.blob.core.windows.net\")"],"metadata":{},"outputs":[],"execution_count":6},{"cell_type":"markdown","source":["## Mount a Blob storage container or a folder inside a container"],"metadata":{}},{"cell_type":"code","source":["# Mount a Blob storage container or a folder inside a container\n# dbutils.fs.mount(\n#   source = \"wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>\",\n#   mount_point = \"<mount-point-path>\",\n#   extra_configs = <\"<conf-key>\": \"<conf-value>\">)\n# [note] <mount_point> is a DBFS path and the path must be under /mnt\n\ndbutils.fs.mount(\n  source = \"wasbs://dbdemo01@databrickstore.blob.core.windows.net\",\n  mount_point = \"/mnt/dbdemo01\",\n  extra_configs = {\"fs.azure.account.key.databrickstore.blob.core.windows.net\": \"S1PtMWvUw5If1Z8FMzXAxC7OMw9G5Go8BGCXJ81qpFVYpZ9dpXOnU4zlg0PbldKkbLIbmbv02WoJsgYLGKIfgg==\"})\n"],"metadata":{},"outputs":[],"execution_count":8},{"cell_type":"markdown","source":["## Access files in your container as if they were local files"],"metadata":{}},{"cell_type":"code","source":["# Access files in your container as if they were local files\n# (TEXT) df = spark.read.text(\"/mnt/%s/....\" % <mount-point-path>)\n# (JSON) df = spark.read.json(\"/mnt/%s/....\" % <mount-point-path>)\n\ndf = spark.read.json( \"/mnt/%s/small_radio_json.json\" % \"dbdemo01\" )\n\n# display(df)\ndf.show()"],"metadata":{},"outputs":[],"execution_count":10},{"cell_type":"markdown","source":["## Unmount the blob storage (if needed)"],"metadata":{}},{"cell_type":"code","source":["# unmount (if needed)\n# dbutils.fs.unmount(\"<mount-point-path>\")\n# dbutils.fs.unmount(\"/mnt/dbdemo01\")"],"metadata":{},"outputs":[],"execution_count":12},{"cell_type":"markdown","source":["# Transform data in Azure Databricks"],"metadata":{}},{"cell_type":"markdown","source":["Start by retrieving only the columns firstName, lastName, gender, location, and level from the dataframe you already created."],"metadata":{}},{"cell_type":"code","source":["specificColumnsDf = df.select(\"firstname\", \"lastname\", \"gender\", \"location\", \"level\")\nspecificColumnsDf.show()"],"metadata":{},"outputs":[],"execution_count":15},{"cell_type":"markdown","source":["You can further transform this data to rename the column level to subscription_type."],"metadata":{}},{"cell_type":"code","source":["renamedColumnsDF = specificColumnsDf.withColumnRenamed(\"level\", \"subscription_type\")\nrenamedColumnsDF.show()"],"metadata":{},"outputs":[],"execution_count":17},{"cell_type":"markdown","source":["# Load data into Azure Cosmos DB"],"metadata":{}},{"cell_type":"markdown","source":["Write configuration, then write to Cosmos DB from the renamedColumnsDF DataFrame"],"metadata":{}},{"cell_type":"code","source":["#writeConfig = {\n# \"Endpoint\" : \"https://<cosmosdb-account-name>.documents.azure.com:443/\",\n# \"Masterkey\" : \"<Cosmosdb-master-key-string>\",\n# \"Database\" : \"<database-name>\",\n# \"Collection\" : \"<collection-name>\",\n# \"Upsert\" : \"true\"\n#}\n\n# Write configuration\nwriteConfig = {\n \"Endpoint\" : \"https://dbstreamdemo.documents.azure.com:443/\",\n \"Masterkey\" : \"ekRLXkETPJ93s6XZz4YubZOw1mjSnoO5Bhz1Gk29bVxCbtgtKmiyRz4SogOSxLOGTouXbwlaAHcHOzct4JVwtQ==\",\n \"Database\" : \"etl\",\n \"Collection\" : \"outcol01\",\n \"Upsert\" : \"true\"\n}\n\n# Write to Cosmos DB from the renamedColumnsDF DataFrame\nrenamedColumnsDF.write.format(\"com.microsoft.azure.cosmosdb.spark\").options(**writeConfig).save()"],"metadata":{},"outputs":[],"execution_count":20}],"metadata":{"name":"elt-blob-storage-cosmosdb-python","notebookId":141860019630215},"nbformat":4,"nbformat_minor":0}