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

{"cells":[{"cell_type":"markdown","source":["# ELT Sample: Azure Blob Stroage - Databricks - SQLDW\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 SQL Data Warehouse.\n\n## prerequisites:\n- Azure Blob Storage Account and Containers\n- Databricks Cluster (Spark)\n- Azure SQL Data Warehouse\n\n## Sample data\n- https://github.com/Azure/usql/blob/master/Examples/Samples/Data/json/radiowebsite/small_radio_json.json\n\n## LINKS\n- https://docs.azuredatabricks.net/spark/latest/data-sources/azure/azure-storage.html\n- https://docs.azuredatabricks.net/spark/latest/data-sources/azure/sql-data-warehouse.html\n- [Quickstart: Create an Azure SQL Data Warehouse](https://docs.microsoft.com/en-us/azure/storage/blobs/storage-quickstart-blobs-portal)"],"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":["# Set up an account access key\n# 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==\"})"],"metadata":{},"outputs":[],"execution_count":8},{"cell_type":"code","source":[""],"metadata":{},"outputs":[],"execution_count":9},{"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":11},{"cell_type":"markdown","source":["## Unmount the blob storage (if needed)"],"metadata":{}},{"cell_type":"code","source":["# unmount\n# dbutils.fs.unmount(\"<mount-point-path>\")\n# dbutils.fs.unmount(\"/mnt/dbdemo01\")"],"metadata":{},"outputs":[],"execution_count":13},{"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":16},{"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":18},{"cell_type":"markdown","source":["# Load data into Azure SQL Data Warehouse"],"metadata":{}},{"cell_type":"code","source":["# Apply some transformations to the data, then use the\n# Data Source API to write the data back to another table in SQL DW.\n\n# [note] the SQL date warehouse connector uses Azure Blob Storage as a temporary storage to upload data between Azure Databricks and Azure SQL Data Warehouse.\n\n## SQL Data Warehouse related settings\ndwTable= \"mytable001\"\ndwDatabase = \"sqldwdemo001\"\ndwServer = \"sqldwdemoserver001\" \ndwUser = \"yoichika\"\ndwPass = \"P@ssw0rd____\"\ndwJdbcPort =  \"1433\"\ndwJdbcExtraOptions = \"encrypt=true;trustServerCertificate=true;hostNameInCertificate=*.database.windows.net;loginTimeout=30;\"\nsqlDwUrl = \"jdbc:sqlserver://\" + dwServer + \".database.windows.net:\" + dwJdbcPort + \";database=\" + dwDatabase + \";user=\" + dwUser+\";password=\" + dwPass + \";$dwJdbcExtraOptions\"\nsqlDwUrlSmall = \"jdbc:sqlserver://\" + dwServer + \".database.windows.net:\" + dwJdbcPort + \";database=\" + dwDatabase + \";user=\" + dwUser+\";password=\" + dwPass\n\n\ntempDir = \"wasbs://dbdemo01tmp@databrickstore.blob.core.windows.net/tempDirs\"\n\n#sc._jsc.hadoopConfiguration().set(\n#  \"fs.azure.account.key.<your-storage-account-name>.blob.core.windows.net\",\n#  \"<your-storage-account-access-key>\")\nacntInfo = \"fs.azure.account.key.databrickstore.blob.core.windows.net\"\nsc._jsc.hadoopConfiguration().set(\n  acntInfo, \n  \"S1PtMWvUw5If1Z8FMzXAxC7OMw9G5Go8BGCXJ81qpFVYpZ9dpXOnU4zlg0PbldKkbLIbmbv02WoJsgYLGKIfgg==\")\n\n## Loading transformed dataframe (renamedColumnsDF) into SQLDW\nspark.conf.set(\"spark.sql.parquet.writeLegacyFormat\",\"true\")\n\n## This snippet creates a table called 'dwTable' in the SQL database.\n#df.write \\\n#  .format(\"com.databricks.spark.sqldw\") \\\n#  .option(\"url\", \"jdbc:sqlserver://<the-rest-of-the-connection-string>\") \\\n#  .option(\"forward_spark_azure_storage_credentials\", \"true\") \\\n#  .option(\"dbtable\", \"my_table_in_dw_copy\") \\\n#  .option(\"tempdir\", \"wasbs://<your-container-name>@<your-storage-account-name>.blob.core.windows.net/<your-directory-name>\") \\\n#  .save()\n\nrenamedColumnsDF.write \\\n  .format(\"com.databricks.spark.sqldw\") \\\n  .option(\"url\", sqlDwUrlSmall) \\\n  .option(\"dbtable\", dwTable) \\\n  .option( \"forward_spark_azure_storage_credentials\",\"true\") \\\n  .option(\"tempdir\", tempDir) \\\n  .mode(\"overwrite\") \\\n  .save()"],"metadata":{},"outputs":[],"execution_count":20}],"metadata":{"name":"elt-blob-storage-sqldw-python","notebookId":3679592423530605},"nbformat":4,"nbformat_minor":0}