05-big-data-analysis-with-dask.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "c2135257-9158-4aec-b0b6-6eb57c5953f1",
   "metadata": {},
   "source": [
    "<img src=\"images/dask-logo.svg\" width=\"20%\" align=\"right\"/>\n",
    "\n",
    "# Big data analysis with Dask\n",
    "\n",
    "In this notebook, we'll work some specific computations and learn some best practices for distributed computing.\n",
    "\n",
    "---\n",
    "\n",
    "_Note: There are some \"Wall time ...\" comments added to the cells, these will change depending on your cluster/profiles, etc. So use them only as a rough reference._\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "94a53a63-dbaf-434d-9576-f9da87fc6f21",
   "metadata": {},
   "source": [
    "## Connect to your Dask Cluster\n",
    "\n",
    "Instead of starting a new cluster each time, you can view and connect to existing clusters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "11a49a5c-ac03-48c4-9d76-ac0bc7166a7e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import dask_gateway"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "890d9ef2-f2e0-471a-bf89-1641bc85cb66",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "gateway = dask_gateway.Gateway()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c1c98878-e275-4911-a995-784bc2ea2559",
   "metadata": {},
   "source": [
    "List your previous created clusters:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8ab83533-cfb2-42ae-a7f3-6f6002328878",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "running_clusters = gateway.list_clusters()\n",
    "running_clusters"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e7ca2ad8-eef2-427f-9acc-afab2cd978d3",
   "metadata": {},
   "source": [
    "Connect to the cluster with it's name, the previously selected options will be preserved:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f3d97d03-86a8-4077-addf-28c60cea33e0",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "cluster = gateway.connect(running_clusters[0].name)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4e45fc3b-c71c-456a-8386-dd0c0d5d41f8",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# # If there aren't any active clusters\n",
    "\n",
    "# options = gateway.cluster_options(use_local_defaults=False)\n",
    "# options.profile = \"Medium Worker\"\n",
    "# options.conda_environment = \"global/global-data-of-unusual-size\"\n",
    "\n",
    "# cluster = gateway.new_cluster(options)\n",
    "# cluster.adapt(minimum=5, maximum=10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "15e58a24-f643-4d09-9009-ce0ade72d759",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "30a7e4a1-978f-4022-9a9f-34527c731f5d",
   "metadata": {},
   "source": [
    "Finally, connect a client to the cluster -- so that you can access it from this specific IPython notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7adcf291-e28a-4db1-9092-94af7b53bd09",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "client = cluster.get_client()\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "022927fe-4f22-4c93-958e-a77b721d8916",
   "metadata": {},
   "source": [
    "Open the Dashboard plots: Cluster Map, Progress, Task Stream, and Workers Memory."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e753cc99-b245-4919-9057-64f701c97790",
   "metadata": {},
   "source": [
    "## Read the full dataset\n",
    "\n",
    "We'll work with the Parquet datasets moving forward!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2389300b-27be-4005-9a7d-cb2617d469eb",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import dask.dataframe as dd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a3def880-250e-40f8-afb8-4ea640033785",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf = dd.read_parquet(\"gcs://quansight-datasets/airline-ontime-performance/full_dataset.parquet\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b97dd370-aff3-47ef-b4db-5492ce3ce588",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "94d115e4-806c-4bd6-9e33-f7ad0c5d3929",
   "metadata": {},
   "source": [
    "## Shuffling computations"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4a86bc19-bd3f-4750-ad81-f2babb59003f",
   "metadata": {},
   "source": [
    "### Calculate and plot the number of canceled flights each day\n",
    "\n",
    "Note that the following section will take 5+ minutes to compute."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bb1c2543-9678-4773-8527-2e1e40ed2521",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import hvplot\n",
    "import hvplot.dask"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9c045dee-8572-4b7d-a8bf-416c33295773",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "hvplot.extension('bokeh') # ensure this cell executes before moving on"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7e7865e6-1b62-4e23-847c-c0b0497b9d33",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "ddf.groupby(\"FL_DATE\")[\"CANCELLED\"].count().hvplot()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5357d827-3b00-4e5d-9ced-7c173b7234c8",
   "metadata": {},
   "source": [
    "This takes:\n",
    "\n",
    "1. so long to compute, \n",
    "2. the plot lines look out of place,\n",
    "3. there is so much \"red\", i.e., interaction, in the task stream\n",
    "\n",
    "because there's an internal \"sort\" operation being done.\n",
    "\n",
    "The \"partition\" based workflow in Dask makes certain types of operations very expensive. Operations like sort, set_index, etc., require dataset-wide interactions, called \"shuffling\", making them very difficult to parallelize.\n",
    "\n",
    "It's a good practice to always **minimize shuffling in distributed computations**.\n",
    "\n",
    "The Dask dashboard is very helpful in catching these unexpected performance penalties\n",
    "\n",
    "Red is a universal color in the Dask dashboard, and it always indicated networking costs when there is data transfer between workers. If you hover on the red bars, you'll see they start with `transfer-` or equivalent."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "88620aed-92a4-4035-86cd-e32cc1617926",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Read the sorted dataset\n",
    "\n",
    "If your computes needs some form of sorting or changing index, you should consider doing it once at the beginning and storing/using the sorted dataset.\n",
    "\n",
    "We have sorted and stored the dataset in the same GCP bucket :)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "06bfc12b-ee1f-4e85-907f-620d2150d2ee",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf = dd.read_parquet(\"gcs://quansight-datasets/airline-ontime-performance/sorted/parquet_by_year\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b9c23025-104d-40d4-892f-05755594832f",
   "metadata": {},
   "source": [
    "### 💻 Your turn: Run the previous computation on the sorted dataset and compare them"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cd5c44dd-4c4f-402e-bc52-8975e6027c51",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here. When ready, click on the three dots below for the solution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "29bd3063-fb49-42f4-b6ca-59bb92803566",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "ddf.groupby(\"FL_DATE\")[\"CANCELLED\"].count().hvplot() "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bb2b2f19-24ca-4603-81c5-1551603c6ce0",
   "metadata": {},
   "source": [
    "## Persisting data and intermediates\n",
    "\n",
    "A \"compute\" in Dask creates a pandas/numeric output and bring the final result to your client machine. This is not required in cases where:\n",
    "\n",
    "- Your client machine (where the output will be displayed) doesn't have enough resources to store/display the output, or\n",
    "- You have more computations to do with the data or intermediate results, and keeping them on the workers will optimize your overall workflow"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8ac24cb2-3ac4-4388-be7d-d15916368269",
   "metadata": {},
   "source": [
    "### Data locality\n",
    "\n",
    "Dask tries to assign computations following \"data locality\", where the computation goes to the worker that holds the required data.\n",
    "\n",
    "If you remember, data transfer is one of the slowest parts of a workflow."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ba2c6647-e174-493e-b1c0-11bfb3297505",
   "metadata": {
    "tags": []
   },
   "source": [
    "### 💻 Your turn: Compute the number of departure and arrival delays per day (without persisting)\n",
    "\n",
    "This is partially similar to the previous workflow where you need to groupby `FL_DATE`. Make sure to record how long this takes!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5cafb392-7a59-4c8b-9e66-b257a8de0bcd",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here. When ready, click on the three dots below for the solutions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "61a7d15f-784b-4dcd-bdb7-2fbbe3130d91",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "ddf.groupby(\"FL_DATE\")[\"DEP_DELAY\"].count().compute() # Wall time: 10.5 s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a8fb7ecc-0c1f-48b0-b80d-9586771de5bd",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "ddf.groupby(\"FL_DATE\")[\"ARR_DELAY\"].count().compute() # Wall time: 5.96 s"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8e76079a-f5eb-49e3-aeba-2a1bac4b540b",
   "metadata": {},
   "source": [
    "### Persist\n",
    "\n",
    "Dask allows you to \"persist\" your data or intermediate outputs (as Dask objects) on the workers.\n",
    "\n",
    "This can run in the background and the control is returned to you immediately."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "676c9958-1912-44fb-9fde-e3ea3089eb0c",
   "metadata": {},
   "source": [
    "### Run the same computation with persisting:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2076ed4a-c78b-4a76-b438-48c8d50af5ab",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf_p = ddf[[\"FL_DATE\", \"DEP_DELAY\", \"ARR_DELAY\"]].persist()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9a223f63-3989-4bd9-91b8-c304d17168c7",
   "metadata": {},
   "source": [
    "Open the \"Graph\" dashboard plot for the following computations!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3cdcc902-95a7-4b5b-9415-be5f32c0a780",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf_p"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7986b02d-ac1d-4bfd-9782-32951f86f57d",
   "metadata": {},
   "source": [
    "Notice how you have the control back immediately, and you can continue working with a Dask DataFrame!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b3f1c0f1-62a9-43a3-8cc1-66485f072845",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "ddf_p.groupby(\"FL_DATE\")[\"DEP_DELAY\"].count().compute() # Wall time: 2.68 s"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1fcd121e-e15c-4e61-92b8-bc54b93b76f9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "ddf_p.groupby(\"FL_DATE\")[\"ARR_DELAY\"].count().compute() # Wall time: 2.47 s"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "836c1b7b-0865-4302-b462-944ec835cde6",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Partitioning effectively"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b8bca518-ad57-4c2d-9039-25db737e8b62",
   "metadata": {},
   "source": [
    "Our dataset currently has 1251 partitions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "802dcb35-ddc7-4a10-9368-f9f081c380d0",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf.npartitions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ac46a374-b561-4083-8d12-6096bb3ec93d",
   "metadata": {},
   "source": [
    "You can change the number of partitions with: `ddf.repartition(npartitions=xx)`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3abb7963-8bf1-4a51-83a1-cb5c2c74c9f2",
   "metadata": {
    "tags": []
   },
   "source": [
    "### 💻 Your turn: Compute the unique flights taken each day, comparing the performance with current partitions and ~600 partitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "94b4b862-f454-4770-abcf-94842876eda8",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here. When ready, click on the three dots below for the solutions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "289c80a7-101a-4028-b757-90fed06ffd0e",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf_full = ddf[[\"FL_DATE\", \"OP_UNIQUE_CARRIER\"]].persist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "036495d7-6b85-4c5c-8887-6bde3b69d0d0",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "ddf_full.groupby(\"FL_DATE\").OP_UNIQUE_CARRIER.count().compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a1ee0f45-fd84-4d15-9791-4eb20da201c0",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf_600 = ddf_full.repartition(npartitions=600).persist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f78aa4b5-0ec9-4b2b-9490-210c4e7abecb",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "ddf_600.groupby(\"FL_DATE\").OP_UNIQUE_CARRIER.count().compute() "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7774e405-fcc9-4e3d-a8c5-50de2c8fee3b",
   "metadata": {},
   "source": [
    "Re-partitioning is an expensive operation. Notice all the red bars in the task graph. \n",
    "\n",
    "Therefore, when you store your data, ensure you have the optimal number of partitions depending on your dataset, your computation, number of workers, and worker resources."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "36bdb6e9-e39b-439a-94fd-6f10a23c037d",
   "metadata": {
    "tags": []
   },
   "source": [
    "## `meta` keyword\n",
    "\n",
    "Since Dask evaluates computations lazily, it uses a special `meta` property to keeps track of the output structure of any computation. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4410e5a8-9325-403c-94a4-7fa6a9fb0dd4",
   "metadata": {
    "tags": []
   },
   "source": [
    "### 💻 Your turn: Convert all negative values in DEP_DELAY to zeros"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b32cb6a1-282a-447a-8f1d-44a99151d84f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here. When ready, click on the three dots below for the solutions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7c3ad1c2-4485-4c0d-b308-6c2095a2582e",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "f = lambda x: 0 if x < 0 else x\n",
    "\n",
    "ddf[\"DEP_DELAY\"].apply(f).compute() # UserWarning: You did not provide metadata ..."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9e075f5d-974e-4d0d-93d7-88fa8d25978e",
   "metadata": {},
   "source": [
    "This is a common warning. `f` is a custom function to Dask can not reliable predict the output structure. In such cases, it's best to specify `meta` explicitly."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5ab22164-d236-4a5c-9d25-44abd22fe53b",
   "metadata": {},
   "source": [
    "### Specify `meta`:\n",
    "\n",
    "You can specify `meta` with an empty/sample pandas data structure with the appropriate columns names and data types.\n",
    "\n",
    "Optional, further reading: [Understanding Dask’s meta keyword argument](https://blog.dask.org/2022/08/09/understanding-meta-keyword-argument)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b0efdf26-2e27-4e4e-934a-6d61e9758537",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7f04d40b-c7a7-4ac1-ac1d-2f91ab60947a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "f = lambda x: 0 if x < 0 else x\n",
    "\n",
    "ddf[\"DEP_DELAY\"].apply(f, meta=pd.Series(dtype=\"float64\")).compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c07e5c4f-f9ce-4d17-822f-e98c62bf837d",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "## Next →\n",
    "\n",
    "Let's look at [big data visualizations](./06-big-data-visualization.ipynb)!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3ca4ccf2-e0c7-4ef2-a8a7-bb7b39204015",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "global-global-data-of-unusual-size",
   "language": "python",
   "name": "conda-env-global-global-data-of-unusual-size-py"
  },
  "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.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}