01-intro-to-dask.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "322b6dca-6b7a-4c56-b2f4-27b887684d6d",
   "metadata": {},
   "source": [
    "<img src=\"images/dask-logo.svg\" width=\"20%\" align=\"right\"/>\n",
    "\n",
    "# Introduction to scalable computing with Dask\n",
    "\n",
    "In this notebook, we'll introduce the dataset and some basic principles of scaling with Dask.\n",
    "\n",
    "---"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a466a26d-571e-4239-9720-ac9cbd76178f",
   "metadata": {
    "tags": []
   },
   "source": [
    "**☝🏽 Important note:**\n",
    "\n",
    "Big data analysis always start with a manageable subset of the data, this allows you to:\n",
    "\n",
    "* Explore it with familiar tools like NumPy and pandas, and\n",
    "* Experiment with various computations you wish to do faster.\n",
    "\n",
    "After you have your computations and pipelines are ready, you can focus on scaling up.\n",
    "\n",
    "To make the most of our time here, we will skip this part and jump right to scaling. If you are curious, you can take a look at the full version of the tutorial at [nebari-dev/big-data-tutorial](https://github.com/nebari-dev/big-data-tutorial)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ebe4f6a7",
   "metadata": {},
   "source": [
    "\n",
    "## Introduce dataset: Airline on-time performance data\n",
    "\n",
    "In this tutorial, we will analyze the **[\"airline on-time performance\" dataset](https://www.transtats.bts.gov/Fields.asp?gnoyr_VQ=FGJ) -- a collection of flight records maintained by the U.S. Department of Transportation's Bureau of Transportation Statistics (BTS)**.\n",
    "\n",
    "This dataset provides information about the on-time performance of domestic flights operated by large air carriers in the United States, including flight delays, cancellations, and diversions. It covers flights operated by 23 major airlines and the records from 1987-present day.\n",
    "\n",
    "We will work with data from 2003-2022, which is ~70 GB in size on disk.\n",
    "\n",
    "The data is stored as one CSV file per month for each year:\n",
    "\n",
    "<img src=\"./images/csv-files.png\">"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d70b3412",
   "metadata": {},
   "source": [
    "## Motivation: Need for scale"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dbc3a6fa-e1fe-4aa1-be7f-3ee92225d56c",
   "metadata": {},
   "source": [
    "Libraries like NumPy and pandas as extremely powerful, however they need your data \"in memory\" (i.e., to fit in your local RAM storage).\n",
    "Which means you won't be able to load the full dataset in pandas without the kernel crashing:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1c499a9c-74a1-48d2-a7bc-41659c6b148d",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Note: Your kernel will restart if you execute this cell.\n",
    "# Uncomment the code below to try for yourself. :)\n",
    "\n",
    "# files = [f\"gcs://{f}\" for f in fs.glob(\"quansight-datasets/airline-ontime-performance/csv/*.csv\")]\n",
    "\n",
    "# with open('prep/dtypes.json', 'r') as f:\n",
    "#     dtypes = json.load(f)\n",
    "\n",
    "# df_list = []\n",
    "# for file in files:\n",
    "#     df_temp = pd.read_csv(file, dtype=dtypes)\n",
    "#     df_list.append(df_temp)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ce86abd7-802d-4253-80d8-a9242e62e2fa",
   "metadata": {},
   "source": [
    "## What is Dask?\n",
    "\n",
    "A library for **parallel and distributed computing in Python**.\n",
    "\n",
    "Dask provides a similar API to familiar PyData libraries (for example NumPy, pandas, scikit-learn) buts runs the same computations in a parallel and/or distributed manner."
   ]
  },
  {
   "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": "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": "code",
   "execution_count": null,
   "id": "112c1810-427a-4073-a36e-5ae8b8cb4f9f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "list(ddf.columns)[0:5]"
   ]
  },
  {
   "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 following cell, Dask loads only the metadata information for the DataFrame, but none of the actual values."
   ]
  },
  {
   "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": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "count = ddf.count()\n",
    "count"
   ]
  },
  {
   "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": [
    "count.visualize() # The task graph is displayed and also saved as a file \"mydask.png\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "742570e4-9657-4477-a7d5-4c8320c9d7b3",
   "metadata": {},
   "source": [
    "We can execute this workflow with `.compute()`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "307688e8-c2a1-406b-9195-6310dc5c6a78",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "count.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "10a59289-ea48-48ed-b84a-853554eae57e",
   "metadata": {},
   "source": [
    "**Note:** Some functions like `.head()` and `.set_index()` also trigger an internal 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 `global/global-pydata-global-2023` 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 5 (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": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf[\"DISTANCE\"].max().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c448b45d-8ce5-4dc7-b44f-b9c12755b1c1",
   "metadata": {
    "tags": []
   },
   "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",
    "[Big data analysis with dask](./02-big-data-analysis-with-dask.ipynb)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "global-global-pydata-global-2023",
   "language": "python",
   "name": "conda-env-global-global-pydata-global-2023-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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}