intro.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "a9dd1484",
   "metadata": {},
   "source": [
    "![dask.png](figs/dask.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cf982502",
   "metadata": {},
   "source": [
    "### Dask installation:\n",
    "`` pip install \"dask[complete]\" pyarrow s3fs graphviz``"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5afc185c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import time\n",
    "import os\n",
    "\n",
    "import dask\n",
    "import dask.dataframe as dd\n",
    "from dask.distributed import LocalCluster, Client\n",
    "import dask.array as da\n",
    "import pandas as pd\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "779c9e81-7489-4369-810b-84c68ed9cf88",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "# The jupyter notebook is launched from your $HOME directory.\n",
    "# Change the working directory to the workshop directory\n",
    "# which was created in your username directory under /scratch/vp91\n",
    "os.chdir(os.path.expandvars(\"/scratch/vp91/$USER/\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "81b224fd",
   "metadata": {},
   "source": [
    "### Dask Collections\n",
    "\n",
    "* **High-level collections**: Mimic NumPy, lists, and pandas but can operate in parallel on datasets that don’t fit into memory \n",
    "    * Array\n",
    "    * DataFrame\n",
    "    * Bag\n",
    "    \n",
    "* **Low-level collections**: Give finer control to build custom parallel and distributed computations\n",
    "    * Delayed\n",
    "    * Futures\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "768ff6b9",
   "metadata": {},
   "source": [
    "# Dask Dataframes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "32855734",
   "metadata": {},
   "source": [
    "![dataframe.png](figs/dataframe.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f28ed45d",
   "metadata": {},
   "source": [
    "* One Dask DataFrame is comprised of many in-memory pandas DataFrames separated along the index. \n",
    "* One operation on a Dask DataFrame triggers many pandas operations on the constituent pandas DataFrames \n",
    "* These operations are mindful of potential parallelism and memory constraints."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2e987f20",
   "metadata": {},
   "outputs": [],
   "source": [
    "!ls distributed-computing-in-dask/data/nycflights/*.csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4555c933",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Read all the csv file into a single Dask dataframe\n",
    "ddf = dd.read_csv(\n",
    "    os.path.join(\"distributed-computing-in-dask/data\", \"nycflights\", \"*.csv\")\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ab6ae552",
   "metadata": {},
   "source": [
    "* dask.dataframe.read_csv only reads in a sample from the beginning of the file\n",
    "* These inferred datatypes are then enforced when reading all partitions\n",
    "* Sometimes, datatypes inferred in the sample can be incorrect. \n",
    "    * The first n rows have no value for CRSElapsedTime (which pandas infers as a float), and later on turn out to be strings (object dtype). \n",
    "\n",
    "* Good practice - specify dtypes directly using the dtype keyword. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "49aa1160",
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf = dd.read_csv(\n",
    "    os.path.join(\"distributed-computing-in-dask/data\", \"nycflights\", \"*.csv\"),\n",
    "    dtype={\"TailNum\": str, \"CRSElapsedTime\": float, \"Cancelled\": bool},\n",
    ")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "25c31314",
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8b751cd6",
   "metadata": {},
   "source": [
    "### Lazy evaluation\n",
    "* Representation of the DataFrame object contains no data \n",
    "* Dask has just done enough to read the start of the first file, and infer the column names and dtypes\n",
    "\n",
    "* Dask **constructs** the logic (called task graph) of your computation immediately\n",
    "* **Evaluates** them only when necessary"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "318b604e",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "ddf.visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4783993f",
   "metadata": {},
   "source": [
    "* Functions like len, head, tail also trigger an evaluation.\n",
    "    * load actual data, (that is, load each file into a pandas DataFrame)\n",
    "    * apply the corresponding functions to each pandas DataFrame (also known as a partition)\n",
    "    * combine the subtotals to give you the final grand total"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4a44577e",
   "metadata": {},
   "outputs": [],
   "source": [
    "len(ddf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e1b5b0e4",
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4058af9c",
   "metadata": {},
   "source": [
    "### Operation on multiple files in Pandas"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5520eda9",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "# find the max value of the DepDelay coulmn in all the 10 dataframes\n",
    "files = os.listdir(os.path.join('distributed-computing-in-dask/data', 'nycflights'))\n",
    "maxes = []\n",
    "\n",
    "for file in files:\n",
    "    df = pd.read_csv(os.path.join('distributed-computing-in-dask/data', 'nycflights', file))\n",
    "    maxes.append(df.DepDelay.max())\n",
    "\n",
    "final_max = max(maxes)\n",
    "print(final_max)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0ea4129b",
   "metadata": {},
   "source": [
    "### Operation on multiple files in Dask"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "90d0806c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# find the max value of the DepDelay coulmn in all the 10 dataframes\n",
    "\n",
    "\n",
    "# This only creates the task graph, it does not execute the operation\n",
    "result = ddf.DepDelay.max()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b5c26d0d",
   "metadata": {},
   "outputs": [],
   "source": [
    "result.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4ebe6330",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "result.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d2004e23",
   "metadata": {},
   "source": [
    "### Excercise: Find the number of flight from each city"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "43b13bb3",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "id": "3aca4399",
   "metadata": {},
   "source": [
    "* We can also combine multiple compute steps into a single instruction\n",
    "* This is usualy more efficient\n",
    "    * Task graphs for both results are merged when calling dask.compute\n",
    "    * shared operations to only be done once instead of twice"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f571807f",
   "metadata": {},
   "outputs": [],
   "source": [
    "non_canceled = ddf[~ddf.Cancelled]\n",
    "mean_delay = non_canceled.DepDelay.mean()\n",
    "std_delay = non_canceled.DepDelay.std()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "81938b4f",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "mean_delay_res = mean_delay.compute()\n",
    "std_delay_res = std_delay.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "151cfa27",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "mean_delay_res, std_delay_res = dask.compute(mean_delay, std_delay)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8e8a0f34",
   "metadata": {},
   "source": [
    "# Dask  Arrays - parallelized numpy"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "83455895",
   "metadata": {},
   "source": [
    "![arrays.png](figs/arrays.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e86a1fa7",
   "metadata": {},
   "source": [
    "* Dask Array implements a subset of the NumPy ndarray interface using **blocked** algorithms\n",
    "* Large array is cut into many small arrays\n",
    "* Large computations are performed by combining many smaller computations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a955fc52",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# NumPy array\n",
    "a_np = np.ones(10)\n",
    "a_np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2c094375",
   "metadata": {},
   "outputs": [],
   "source": [
    "# This is how a blocked operation is done in numpy. We divide the whole ndarray\n",
    "# of size 10 int slices of 2, each of size 5\n",
    "\n",
    "a_np_sum = a_np[:5].sum() + a_np[5:].sum()\n",
    "a_np_sum"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9cc0677c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Dask array\n",
    "\n",
    "# In task ndarray we specify the slices usinh the keyword chunk. \n",
    "# chunk defines the numer of elements in each slice\n",
    "\n",
    "a_da = da.ones(10, chunks=5)\n",
    "a_da"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4aea340e",
   "metadata": {},
   "outputs": [],
   "source": [
    "a_da_sum = a_da.sum()\n",
    "a_da_sum"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e7d6ad7b",
   "metadata": {},
   "outputs": [],
   "source": [
    "a_da_sum.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d0918195",
   "metadata": {},
   "outputs": [],
   "source": [
    "a_da_sum.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f394b00d",
   "metadata": {},
   "source": [
    "* Dask can also find an optimal chunk by itself\n",
    "* If your chunks are too small\n",
    "    * the amount of actual work done by every task is very tiny\n",
    "    * the overhead of coordinating all these tasks results in a very inefficient process\n",
    "* If your chunks are too big\n",
    "    * you will likely run out of memory\n",
    "    * data will have to be moved to the disk \n",
    "    * this will lead to performance decrements"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8a1ca08a",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "xd = da.random.normal(10, 0.1, size=(30_000, 30_000), chunks=(3000, 3000)) # We specify the chunk\n",
    "yd = xd.mean(axis=0)\n",
    "yd.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e541ba04",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "xd = da.random.normal(10, 0.1, size=(30_000, 30_000)) # Dask finds the chunk\n",
    "yd = xd.mean(axis=0)\n",
    "yd.compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2c63ac1d",
   "metadata": {},
   "outputs": [],
   "source": [
    "xd.chunksize"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "20aa8ddf",
   "metadata": {},
   "source": [
    "# Delayed decorator"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "86390e38",
   "metadata": {},
   "source": [
    "* A Block of code can have operations that can happen in parallel\n",
    "* Normally in python these operation will happen sequentially\n",
    "    * Or the user will identify the parallel section and write parallel codes\n",
    "* The Dask **delayed** function decorates your functions so that they operate lazily \n",
    "* Dask will defer execution of the function, placing the function and its arguments into a task graph\n",
    "* Dask will then identify oppurtunities for parallelism in the task graph\n",
    "* The Dask schedulers will exploit this parallelism, generally improving performance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ad9e992c",
   "metadata": {},
   "outputs": [],
   "source": [
    "@dask.delayed\n",
    "def inc(x):\n",
    "    time.sleep(1)\n",
    "    return x + 1\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "22518674",
   "metadata": {},
   "outputs": [],
   "source": [
    "@dask.delayed\n",
    "def add(x, y):\n",
    "    time.sleep(1)\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "aa3fc2ec",
   "metadata": {},
   "outputs": [],
   "source": [
    "# As the two increments are independent of each other, we can run them in parallel\n",
    "\n",
    "x = inc(1)\n",
    "y = inc(2)\n",
    "z = add(x, y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9236cc4c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Here Z is a delayed object\n",
    "\n",
    "z.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cd08d0d0",
   "metadata": {},
   "outputs": [],
   "source": [
    "z.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0527c16a",
   "metadata": {},
   "source": [
    "# Dask future"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c52514cf",
   "metadata": {},
   "source": [
    "* we can submit individual functions for evaluation\n",
    "* The call returns immediately, giving one or more future\n",
    "    * whose status begins as “pending”\n",
    "    * later becomes “finished”\n",
    "* There is no **blocking** of the local Python session.\n",
    "\n",
    "* Difference between futures and delayed\n",
    "    * delayed is lazy (it just constructs a graph) \n",
    "    * futures are eager. \n",
    "    * With futures, as soon as the inputs are available and there is compute available, the computation starts"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ad2cd975",
   "metadata": {},
   "outputs": [],
   "source": [
    "client = Client(n_workers=4)\n",
    "\n",
    "def inc(x):\n",
    "    time.sleep(1)\n",
    "    return x + 1\n",
    "\n",
    "\n",
    "def double(x):\n",
    "    sleep(2)\n",
    "    return 2 * x\n",
    "\n",
    "\n",
    "def add(x, y):\n",
    "    time.sleep(1)\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "50800b1c",
   "metadata": {},
   "outputs": [],
   "source": [
    "future = client.submit(inc, 1)  # returns immediately with pending future\n",
    "future"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f79a0979",
   "metadata": {},
   "source": [
    "#### If we check the future after a few seconds we can see that it is complete"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3af0a636",
   "metadata": {},
   "outputs": [],
   "source": [
    "future"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4429a996",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "future.result()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ed10d6f7",
   "metadata": {},
   "outputs": [],
   "source": [
    "client.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a15fad62",
   "metadata": {},
   "source": [
    "# Distributed Dask "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "15a412a9",
   "metadata": {},
   "source": [
    "![dask_cluster.png](figs/dask_cluster.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b648f4e7",
   "metadata": {},
   "source": [
    "* Dask can work on a clusteer\n",
    "* We have been using the distributed scheduler for our work, but just on a single machine.\n",
    "* When we instantiate a Client() object with no arguments it will attempt to locate a Dask cluster\n",
    "    * It will check your local Dask config and environment variables to see if connection information has been specified\n",
    "    * If not it will create an instance of LocalCluster and use that"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f1173b47",
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster = LocalCluster()\n",
    "cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "88ebb5fd",
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster.get_logs()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "407d92a3",
   "metadata": {},
   "outputs": [],
   "source": [
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "46ff1dbb",
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "del client, cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "34c633ea-72a2-4bd4-bb55-68fe3d52e84f",
   "metadata": {},
   "source": [
    "## Compute Vs Persist"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2f15afca-7308-46a2-9116-b38b69de5fb1",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = dask.datasets.timeseries()\n",
    "df.npartitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1f06b599-3349-47a3-85df-7a198fc72b6d",
   "metadata": {},
   "outputs": [],
   "source": [
    "type(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7ba8b910-6f1b-4c3a-8d23-5056ee9d341d",
   "metadata": {},
   "outputs": [],
   "source": [
    "computed_df = df.compute()\n",
    "type(computed_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9b01cf34-cd0e-41c7-92bf-3dca4d51d861",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_persist = df.persist()\n",
    "type(df_persist)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f2772e95-ec32-4446-8311-026a22c71367",
   "metadata": {},
   "outputs": [],
   "source": [
    "df_persist.npartitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4f0c9364-3fac-4df4-8b56-d0eda67ae79c",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9fe9a34f-ef9f-41bb-ac5f-e0c3111ce8a8",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "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.11.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}