tutorials/using-dask/1.intro-to-dask.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Introduction to Dask\n",
    "\n",
    "In this notebook, we'll learn how to use [Dask](https://dask.org) for reading and processing data from Azure."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Install required packages"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install --upgrade dask distributed dask-sql bokeh adlfs fsspec fastparquet pyarrow python-snappy lz4 \"pandas>=1.2.0\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Get AML Workspace\n",
    "\n",
    "You can use the AML workspace to retrieve datastores and keyvaults for accessing data credentials securely."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from azureml.core import Workspace\n",
    "\n",
    "ws = Workspace.from_config()\n",
    "ws"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create a distributed client\n",
    "\n",
    "The [client](https://distributed.dask.org/en/latest/client.html) is the primary entrypoint for parallel processing with Dask. Calling it without inputs will create a local distributed scheduler, utilizing all the CPUs and cores on your machine. This can be useful for faster processing of larger in memory dataframes, or even computations on out of memory (OOM) data. \n",
    "\n",
    "When your local machine isn't powerful enough, you can provision a larger VM in Azure - the M series has 100+ CPUs and TBs of RAM. If this still isn't powerful enough, you can create a distributed Dask cluster on most hardware - see [the Dask setup guide](https://docs.dask.org/en/latest/setup.html) for details.\n",
    "\n",
    "If you still need acceleration, [RAPIDSAI](https://github.com/rapidsai) further extends the PyData APIs on GPUs.\n",
    "\n",
    "**Make sure you check out the dashboard!**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from distributed import Client\n",
    "\n",
    "# initialize local client\n",
    "c = Client()\n",
    "\n",
    "# print Python objects\n",
    "print(c)\n",
    "print(c.dashboard_link)\n",
    "\n",
    "# print notebook widget widget\n",
    "c"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Reading cloud data\n",
    "\n",
    "Reading data from the cloud is as easy! Python implements various cloud protocols, including ``az`` for Blob and ADLSv2 and ``adl`` for ADLSv1.\n",
    "\n",
    "### Public Data\n",
    "\n",
    "Public data can simply be read via ``https``.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "account_name = \"azuremlexamples\"\n",
    "container_name = \"datasets\"\n",
    "\n",
    "storage_options = {\"account_name\": account_name}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_uri = f\"https://{account_name}.blob.core.windows.net/{container_name}/iris.csv\"\n",
    "data_uri"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "df = pd.read_csv(data_uri)\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "alternatively, we can use the ``az`` protocol and pass in ``storage_options``:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_uri = f\"az://{container_name}/iris.csv\"\n",
    "data_uri"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "df = pd.read_csv(data_uri, storage_options=storage_options)\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Private Data \n",
    "\n",
    "Passing in storage options allows for reading private data. For instance, you can easily retrieve the information from an Azure ML Datastore:\n",
    "\n",
    "```python\n",
    "from azureml.core import Workspace\n",
    "\n",
    "ws = Workspace.from_config()\n",
    "ds = ws.get_default_datastore() # ws.datastores[\"my-datastore-name\"]\n",
    "\n",
    "container_name = ds.container_name\n",
    "storage_options = {\n",
    "    \"account_name\": ds.account_name,\n",
    "    \"account_key\": ds.account_key,\n",
    "}\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from adlfs import AzureBlobFileSystem\n",
    "\n",
    "ds = ws.get_default_datastore()\n",
    "\n",
    "container_name = ds.container_name\n",
    "storage_options = {\"account_name\": ds.account_name, \"account_key\": ds.account_key}\n",
    "\n",
    "fs = AzureBlobFileSystem(**storage_options)\n",
    "fs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs.ls(f\"{container_name}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pythonic Filesystem\n",
    "\n",
    "In the previous section, we used [ADLFS](https://github.com/dask/adlfs) to initialize a Pythonic filesystem and perform operations.\n",
    "\n",
    "The below cell demonstrate some basic operations to raed and manipulate data in Python."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "color = \"green\"\n",
    "container_name = \"nyctlc\"\n",
    "storage_options = {\"account_name\": \"azureopendatastorage\"}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs = AzureBlobFileSystem(**storage_options)\n",
    "fs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs.ls(f\"{container_name}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs.ls(f\"{container_name}/{color}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fs.ls(f\"{container_name}/{color}/puYear=2016/\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "files = fs.glob(f\"{container_name}/{color}/puYear=2016/puMonth=12/*.parquet\")\n",
    "files = [f\"az://{file}\" for file in files]\n",
    "files[-5:]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import dask.dataframe as dd\n",
    "\n",
    "ddf = (\n",
    "    dd.read_parquet(files, storage_options=storage_options)\n",
    "    .repartition(npartitions=8)\n",
    "    .persist()\n",
    ")\n",
    "ddf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "len(ddf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "len(ddf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import matplotlib.pyplot as plt\n",
    "\n",
    "plt.style.use(\"dark_background\")\n",
    "\n",
    "ddf[\"tipAmount\"].compute().hist(\n",
    "    figsize=(16, 8),\n",
    "    bins=256,\n",
    "    range=(0.1, 20),\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = ddf.compute()\n",
    "df.info()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "df.describe()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "gbs = round(df.memory_usage(index=True, deep=True).sum() / 1e9, 2)\n",
    "print(f\"df is {gbs} GBs\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "gbs = round(ddf.memory_usage(index=True, deep=True).sum().compute() / 1e9, 2)\n",
    "print(f\"ddf is {gbs} GBs\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.8",
   "language": "python",
   "name": "python3.8"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.5-final"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}