03-intro-to-dask.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "322b6dca-6b7a-4c56-b2f4-27b887684d6d",
   "metadata": {},
   "source": [
    "# Introduction to scalable computing with Dask\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ce86abd7-802d-4253-80d8-a9242e62e2fa",
   "metadata": {},
   "source": [
    "## What is Dask?\n",
    "\n",
    "<img src=\"images/dask-logo.svg\" width=\"20%\" align=\"right\"/>\n",
    "\n",
    "A library for **parallel and distributed computing in Python**.\n",
    "\n",
    "Traditionally PyData libraries were designed for linear workflows (for example NumPy, pandas, scikit-learn),\n",
    "Dask provides a similar API to run the same computations in parallel."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "117446ca-f183-4354-925c-c93bff03e989",
   "metadata": {},
   "source": [
    "## Parallel computing\n",
    "\n",
    "Computing parts of a workflow simultaneously. Typically, we use this term to describe single-machine parallelism, where your computation can be run simultaneously on various cores of a machine while sharing the same memory (RAM)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9c6162df-0082-401c-9da0-da3336801aea",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Dask DataFrame API\n",
    "\n",
    "Dask has a few different APIs to parallelize different tools/activities. We will primarily cover Dask's DataFrame API, which parallelizes pandas, in this tutorial.\n",
    "\n",
    "The idea is to provide a familiar interface to pandas, but leverage parallelism under-the-hood."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0045d6ac-fc68-42de-a5eb-b557e5b69359",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import dask.dataframe as dd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c2343b3a-1fa1-46ec-ab34-25347e2d3091",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf = dd.read_csv(\"gcs://quansight-datasets/airline-ontime-performance/csv/*ber_2020.csv\") # September, October, November, December - 2020"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6293a01c-be14-4199-8b12-50b95110e1eb",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "994494da-c7ff-4cf4-b442-094833d10141",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Lazy evaluation\n",
    "\n",
    "Dask evaluates your computations lazily, this is what allows Dask to \"scale\" your computations. This means, Dask only creates the \"logic\" of your computation eagerly, i.e., what are the independent tasks that can be executed in parallel, what does that dependency tree (called \"task graph\" in Dask) look like.\n",
    "\n",
    "In the previous cell, Dask has loaded only the metadata information for the DataFrame, but none of the actual values."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c6b7cb43-18ab-4730-a57d-90ce68f59d13",
   "metadata": {},
   "source": [
    "You can use `.compute()` to run/execute the computations:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0a1b3810-af5a-4f90-a48a-05e7ffa7cd3f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf.OP_UNIQUE_CARRIER.unique().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "10a59289-ea48-48ed-b84a-853554eae57e",
   "metadata": {},
   "source": [
    "Some commands like `.head()` also trigger an internal compute, so you get the "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ea143667-329e-4011-9ce9-b994ad8ae304",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf.head() # ValueError: Mismatched dtypes found in `pd.read_csv`/`pd.read_table`."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8922daba-8cd0-4057-af6f-7d3afa51a4c6",
   "metadata": {},
   "source": [
    "### Specify `dtypes`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ed649947-8f90-4c8b-ad1b-01c6b8718271",
   "metadata": {},
   "source": [
    "The lazy behavior of Dask means it infers the datatypes using minimal information -- for CSV files, Dask uses the first row.\n",
    "\n",
    "This behavior is different from pandas, which loads the entire dataset and then infers datatypes.\n",
    "\n",
    "Hence, in Dask and distributed computing in general tt's a good practice to provide explicit dtypes, especially if you use CSV files. You can do this by exporting the dtypes of a subset of data, We've already prepared the dtypes for this tutorial:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "69f29d63-a01b-4720-bcdc-10c55b603c41",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "with open('prep/dtypes.json', 'r') as f:\n",
    "    dtypes = json.load(f)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "02364589-2981-455e-a533-227caa12f65b",
   "metadata": {},
   "source": [
    "(Optional: You can take a look at `prep/dtypes.json`)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1d7977a8-6855-4c72-beeb-55d1eb458b21",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf = dd.read_csv(\"gcs://quansight-datasets/airline-ontime-performance/csv/*ber_2020.csv\", dtype=dtypes)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "185829d8-69cb-47cb-a9c9-eb887c9b6d97",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf.head() # No warnings or errors :)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d257c06c-2b63-4df8-8223-390c4e5a6aec",
   "metadata": {},
   "source": [
    "When we do computations, Dask keeps track of the parallel logic based on what it expects the output structure for each operation to look like."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3de8222a-ff74-429a-bb09-59d947b0ac56",
   "metadata": {
    "scrolled": true,
    "tags": []
   },
   "outputs": [],
   "source": [
    "add = ddf.DISTANCE.sum()\n",
    "add"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f49532b3-86f8-4f30-b3ef-4c9135b9970c",
   "metadata": {},
   "source": [
    "In the following task graph, everything in the same horizontal layer will be executed in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "23defc83-e9c1-456b-88b2-f18b42da01ee",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "add.visualize() # In case this doesn't display the task graph eagerly, open \"mydask.png\" file created"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "742570e4-9657-4477-a7d5-4c8320c9d7b3",
   "metadata": {},
   "source": [
    "We can executes this workflow with `compute()`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "307688e8-c2a1-406b-9195-6310dc5c6a78",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "add.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aaeef1df-b3f4-404c-844d-2b73c2dbf5cb",
   "metadata": {},
   "source": [
    "### Partitions\n",
    "\n",
    "Internally, Dask DataFrame is a collection of pandas DataFrames (these are actual pandas DataFrames internally as well!):\n",
    "\n",
    "<img src=\"./images/dask-dataframe.svg\" width=\"30%\"/>\n",
    "\n",
    "where each pandas DataFrame is called a \"partition\".\n",
    "\n",
    "Your Dask computations will be run on all the individual pandas DataFrames in parallel, and then combined as necessary."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "373f3ed5-ba0e-4fa9-bea5-dd2eba037875",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf.npartitions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fedcf25d-2a2c-43f0-ae85-e4c3fc4e77b3",
   "metadata": {},
   "source": [
    "Dask selects an adequate number of partitions based on your dataset and resource limits. If you use partitioned data formats, like Parquet (we'll learn more later!), Dask will preserve the partitions while reading data."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ce402d85-e354-4b57-9484-4de37bded430",
   "metadata": {},
   "source": [
    "## Distributed computing\n",
    "\n",
    "We can also leverage parallel computation on several different machines (workers) with their own processors and memory.\n",
    "The different machines can interact to share data, and a central machine (scheduler) manages all the interactions.\n",
    "Distributed computing refers to this system/model of computations.\n",
    "\n",
    "These different machines can be located anywhere, on your local in-house network or in data centers around the world."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d250712f-e464-425a-b24b-8be29dca1672",
   "metadata": {
    "tags": []
   },
   "source": [
    "<img src=\"images/distributed-overview.png\" width=\"50%\"/>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e7f15212-5e19-426c-9cdb-ff6954ea5358",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Dask Gateway\n",
    "\n",
    "Dask Gateway is a library to manage Dask clusters on the cloud. The platform you're on (Nebari) has a Dask Gateway and ee'll create clusters using Google Cloud Provider machines in this tutorial.\n",
    "\n",
    "<img src=\"images/gateway-architecture.svg\" width=\"50%\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6bb91fa0-e55e-44a0-9acd-eaddab915242",
   "metadata": {},
   "source": [
    "First, import the library and create a new Gateway instance:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6d802c6e-db6f-43b4-9f63-c99f0ca173e0",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import dask_gateway"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "574eb96f-e540-4b6d-ac01-2daaa42700c4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "gateway = dask_gateway.Gateway()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "76ebd754-a233-47ad-a5ba-fa270c5b06ac",
   "metadata": {},
   "source": [
    "Then set how your workers need to be configured, and make sure you select:\n",
    "* the same `scipy/scipy-data-of-unusual-size` environment as your current notebook (aka, client), and\n",
    "* \"Medium Worker\" cluster profile."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "84fcbf73-44ad-489c-9921-f7d43865dd85",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "options = gateway.cluster_options(use_local_defaults=False)\n",
    "options"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "85ef1e2f-dc6d-48ef-aebc-37321b61f605",
   "metadata": {},
   "source": [
    "### Manual vs adaptive scaling\n",
    "\n",
    "You can specify the exact number of machines required in your cluster, and Dask will spin all of them up at the beginning. We call this approach \"Manual scaling\".\n",
    "\n",
    "Dask Gateway also has a very useful \"adaptive scaling\" feature. This allows Dask to spin up new machines as your workflow/computation needs it, and then shut idle workers down safely after the computation is complete, until the next computation is triggered. Adaptive scaling can help manage costs when you have large compute requirements.\n",
    "\n",
    "In the widget below, select adaptive scaling, set 3 (min) and 10 (max), and click \"Adapt\":"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "797f30c3-a80b-491a-b6a3-11911b78e72b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "cluster = gateway.new_cluster(options)\n",
    "cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "671a94fc-19ef-45ac-bffa-4f1ae2cf2444",
   "metadata": {},
   "source": [
    "Finally, you can connect this cluster of machines to this IPython notebook using a client:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a3e4548d-2da9-4020-a89b-c5b3b8776c9e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "client = cluster.get_client()\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "af27c956-65c6-44c7-8cfd-0955f0d83757",
   "metadata": {},
   "source": [
    "You slowly start getting new machines and you can see the Workers, Threads, and Memory increase in the \"GatewayCluster\" widget."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8106f5eb-74d1-440b-a43d-b47ab48f8da2",
   "metadata": {},
   "source": [
    "## Dask Dashboard\n",
    "\n",
    "The Client widget displays a link to a dashboard:\n",
    "* Click on it, and a new Keycloak sign-up page should open\n",
    "* Login with the email and password you used to register\n",
    "* The dashboard opens in the browser window\n",
    "\n",
    "You will need to login only once, you should be able to access the Dashboard directly if you click on the link next time. :)\n",
    "\n",
    "You can also access these plots within JupyterLab:\n",
    "* Click on the Dask logo in the left sidebar\n",
    "* Click on the magnifying glass icon, the dashboard should connect automatically and display available plots\n",
    "* Open: \"Cluster map\", \"task stream\", and \"progress\" plots\n",
    "* Re-arrange the plots in your JupyterLab interface to see them all together\n",
    "\n",
    "Your instructor will talk about each plot as you work on the following computations!"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bfcbcf93-b87a-49c4-a8f8-203ddc6f6658",
   "metadata": {},
   "source": [
    "## A quick computation"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6cc25323-7706-4d2a-8465-f13c2587f6d2",
   "metadata": {},
   "source": [
    "### 💻 Your turn: Compute the longest flight (distance) across the dataset\n",
    "\n",
    "Make sure to look at the dashboard plots :)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "08f60f5d-21e4-4d19-aaab-57bff40113de",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Your code here. When ready, click on the three dots below for the solution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0e5688f3-ac4e-44ed-9a92-dbfaff904c1e",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf[\"DISTANCE\"].max()\n",
    "ddf[\"DISTANCE\"].max().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c448b45d-8ce5-4dc7-b44f-b9c12755b1c1",
   "metadata": {},
   "source": [
    "## Ensure cluster shutdown\n",
    "\n",
    "Idling clusters can quickly add up to costs, so make sure to always shutdown your clusters after completing your work."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7eed29cc-4740-48de-b936-ed48795836b2",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "cluster.shutdown()\n",
    "client.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c8d51abc-2f51-42c1-86e1-5555d85c6d8a",
   "metadata": {},
   "source": [
    "---\n",
    "\n",
    "## Next →\n",
    "\n",
    "[Storage formats](./04-storage-formats.ipynb)!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b0d9c69f-5e9c-4c5a-9ee6-078ed7e79190",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "scipy-scipy-data-of-unusual-size",
   "language": "python",
   "name": "conda-env-scipy-scipy-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.14"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}