sdk/python/data-wrangling/interactive_data_wrangling.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Interactive Data Wrangling using Apache Spark in Azure Machine Learning\n",
    "Before executing these sample codes in an Azure Machine Learning Notebook, select **Serverless Spark Compute** under **Azure Machine Learning Serverless Spark** or select an attached Synapse Spark pool under **Synapse Spark pools** from the **Compute** selection menu. It is highly recommened to follow the documentation page: [Interactive data wrangling with Apache Spark in Azure Machine Learning](https://learn.microsoft.com/azure/machine-learning/interactive-data-wrangling-with-apache-spark-azure-ml) for more details related to the code samples provided in this notebook."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Access and wrangle Azure Blob storage data using Access Key"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, Set the access key as configuration property `fs.azure.account.key.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "sc = SparkSession.builder.getOrCreate()\n",
    "token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary\n",
    "access_key = token_library.getSecret(\"<KEY_VAULT_NAME>\", \"<ACCESS_KEY_SECRET_NAME>\")\n",
    "sc._jsc.hadoopConfiguration().set(\n",
    "    \"fs.azure.account.key.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net\", access_key\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Access data using `wasbs://` URI and perform data wrangling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "\n",
    "df = pd.read_csv(\n",
    "    \"wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/titanic.csv\",\n",
    "    index_col=\"PassengerId\",\n",
    ")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(\n",
    "    \"wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/wrangled\",\n",
    "    index_col=\"PassengerId\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Access and wrangle Azure Blob storage data using SAS token"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, set the SAS token as configuration property `fs.azure.sas.<BLOB_CONTAINER_NAME>.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "sc = SparkSession.builder.getOrCreate()\n",
    "token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary\n",
    "sas_token = token_library.getSecret(\"<KEY_VAULT_NAME>\", \"<SAS_TOKEN_SECRET_NAME>\")\n",
    "sc._jsc.hadoopConfiguration().set(\n",
    "    \"fs.azure.sas.<BLOB_CONTAINER_NAME>.<STORAGE_ACCOUNT_NAME>.blob.core.windows.net\",\n",
    "    sas_token,\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Access data using `wasbs://` URI and perform data wrangling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "\n",
    "df = pd.read_csv(\n",
    "    \"wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/titanic.csv\",\n",
    "    index_col=\"PassengerId\",\n",
    ")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(\n",
    "    \"wasbs://<BLOB_CONTAINER_NAME>@<STORAGE_ACCOUNT_NAME>.blob.core.windows.net/data/wrangled\",\n",
    "    index_col=\"PassengerId\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Access and wrangle ADLS Gen 2 data using User Identity passthrough"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- To enable read and write access, assign **Contributor** and **Storage Blob Data Contributor** roles to the user identity.\n",
    "- Access data using `abfss://` URI and perform data wrangling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "\n",
    "df = pd.read_csv(\n",
    "    \"abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/titanic.csv\",\n",
    "    index_col=\"PassengerId\",\n",
    ")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(\n",
    "    \"abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/wrangled\",\n",
    "    index_col=\"PassengerId\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Access and wrangle ADLS Gen 2 data using Service Principal"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- To enable read and write access, assign **Contributor** and **Storage Blob Data Contributor** roles to the user identity.\n",
    "- Set configuration properties as follows:\n",
    "    - Client ID property: `fs.azure.account.oauth2.client.id.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net`\n",
    "    - Client secret property: `fs.azure.account.oauth2.client.secret.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net`\n",
    "    - Tenant ID property: `fs.azure.account.oauth2.client.endpoint.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net`\n",
    "    - Tenant ID value: `https://login.microsoftonline.com/<TENANT_ID>/oauth2/token`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "sc = SparkSession.builder.getOrCreate()\n",
    "token_library = sc._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary\n",
    "\n",
    "# Set up service principal tenant ID, client ID and secret from Azure Key Vault\n",
    "client_id = token_library.getSecret(\"<KEY_VAULT_NAME>\", \"<CLIENT_ID_SECRET_NAME>\")\n",
    "tenant_id = token_library.getSecret(\"<KEY_VAULT_NAME>\", \"<TENANT_ID_SECRET_NAME>\")\n",
    "client_secret = token_library.getSecret(\"<KEY_VAULT_NAME>\", \"<CLIENT_SECRET_NAME>\")\n",
    "\n",
    "# Set up service principal which has access of the data\n",
    "sc._jsc.hadoopConfiguration().set(\n",
    "    \"fs.azure.account.auth.type.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net\", \"OAuth\"\n",
    ")\n",
    "sc._jsc.hadoopConfiguration().set(\n",
    "    \"fs.azure.account.oauth.provider.type.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net\",\n",
    "    \"org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider\",\n",
    ")\n",
    "sc._jsc.hadoopConfiguration().set(\n",
    "    \"fs.azure.account.oauth2.client.id.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net\",\n",
    "    client_id,\n",
    ")\n",
    "sc._jsc.hadoopConfiguration().set(\n",
    "    \"fs.azure.account.oauth2.client.secret.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net\",\n",
    "    client_secret,\n",
    ")\n",
    "sc._jsc.hadoopConfiguration().set(\n",
    "    \"fs.azure.account.oauth2.client.endpoint.<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net\",\n",
    "    \"https://login.microsoftonline.com/\" + tenant_id + \"/oauth2/token\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Access data using `abfss://` URI and perform data wrangling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "\n",
    "df = pd.read_csv(\n",
    "    \"abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/titanic.csv\",\n",
    "    index_col=\"PassengerId\",\n",
    ")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(\n",
    "    \"abfss://<FILE_SYSTEM_NAME>@<STORAGE_ACCOUNT_NAME>.dfs.core.windows.net/data/wrangled\",\n",
    "    index_col=\"PassengerId\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Access and wrangle data using credentialed AzureML Blob Datastore"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Access data using `azureml://` URI and perform data wrangling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "\n",
    "df = pd.read_csv(\n",
    "    \"azureml://datastores/workspaceblobstore/paths/data/titanic.csv\",\n",
    "    index_col=\"PassengerId\",\n",
    ")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(\n",
    "    \"azureml://datastores/workspaceblobstore/paths/data/wrangled\",\n",
    "    index_col=\"PassengerId\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Access and wrangle data using credentialless AzureML Blob Datastore"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- To enable read and write access, assign **Contributor** and **Storage Blob Data Contributor** roles to the user identity on the Azure Blob storage account that the datastore points to.\n",
    "- Access data using `azureml://` URI and perform data wrangling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "\n",
    "df = pd.read_csv(\n",
    "    \"azureml://datastores/credlessblobdatastore/paths/data/titanic.csv\",\n",
    "    index_col=\"PassengerId\",\n",
    ")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(\n",
    "    \"azureml://datastores/credlessblobdatastore/paths/data/wrangled\",\n",
    "    index_col=\"PassengerId\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Access credentialed AzureML ADLS Gen 2 Datastore"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- To enable read and write access, assign **Contributor** and **Storage Blob Data Contributor** roles to the service principal used by datastore.\n",
    "- Access data using `azureml://` URI and perform data wrangling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "\n",
    "df = pd.read_csv(\n",
    "    \"azureml://datastores/adlsg2datastore/paths/data/titanic.csv\",\n",
    "    index_col=\"PassengerId\",\n",
    ")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(\n",
    "    \"azureml://datastores/credadlsg2datastore/paths/data/wrangled\",\n",
    "    index_col=\"PassengerId\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Access and wrangle data using credentialless AzureML ADLS Gen 2 Datastore"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- To enable read and write access, assign **Contributor** and **Storage Blob Data Contributor** roles to the user identity on the Azure Data Lake Storage (ADLS) Gen 2 storage account that the datastore points to.\n",
    "- Access data using `azureml://` URI and perform data wrangling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import pyspark.pandas as pd\n",
    "from pyspark.ml.feature import Imputer\n",
    "\n",
    "df = pd.read_csv(\n",
    "    \"azureml://datastores/credlessadlsg2datastore/paths/data/titanic.csv\",\n",
    "    index_col=\"PassengerId\",\n",
    ")\n",
    "imputer = Imputer(inputCols=[\"Age\"], outputCol=\"Age\").setStrategy(\n",
    "    \"mean\"\n",
    ")  # Replace missing values in Age column with the mean value\n",
    "df.fillna(\n",
    "    value={\"Cabin\": \"None\"}, inplace=True\n",
    ")  # Fill Cabin column with value \"None\" if missing\n",
    "df.dropna(inplace=True)  # Drop the rows which still have any missing value\n",
    "df.to_csv(\n",
    "    \"azureml://datastores/credlessadlsg2datastore/paths/data/wrangled\",\n",
    "    index_col=\"PassengerId\",\n",
    ")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "source": [
    "### Access mounted File Share"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Access data on mounted File share by constructing absolute path."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "outputs_hidden": false,
     "source_hidden": false
    },
    "nteract": {
     "transient": {
      "deleting": false
     }
    }
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import pyspark.pandas as pd\n",
    "\n",
    "abspath = os.path.abspath(\".\")\n",
    "file = \"file://\" + abspath + \"/Users/<USER>/data/titanic.csv\"\n",
    "print(file)\n",
    "df = pd.read_csv(file)\n",
    "df.head()"
   ]
  }
 ],
 "metadata": {
  "kernel_info": {
   "name": "synapse_pyspark"
  },
  "kernelspec": {
   "display_name": "Python 3.10 - SDK V2",
   "language": "python",
   "name": "python310-sdkv2"
  },
  "language_info": {
   "codemirror_mode": "ipython",
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython",
   "version": "3.7.10"
  },
  "microsoft": {
   "language": "python"
  },
  "nteract": {
   "version": "nteract-front-end@1.0.0"
  },
  "vscode": {
   "interpreter": {
    "hash": "6aeff17a1aa7735c2f7cb3a6d691fe1b4d4c3b8d2d650f644ad0f24e1b8e3f3f"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}