02_dataframe.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8f20d516",
   "metadata": {
    "tags": []
   },
   "source": [
    "<img src=\"https://raw.githubusercontent.com/dask/dask/main/docs/source/images/dask_horizontal.svg\"\n",
    "     width=\"60%\"\n",
    "     alt=\"Dask logo\\\" />\n",
    "\n",
    "# Process Tabular Data with Dask DataFrame\n",
    "In this notebook we will learn about the [Dask DataFrame](https://docs.dask.org/en/latest/dataframe.html), a tabular DataFrame interface based on pandas that will automatically build parallel computations."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "595e457f",
   "metadata": {
    "tags": []
   },
   "source": [
    "## When to use Dask DataFrames\n",
    "\n",
    "Pandas is great for tabular datasets that fit in memory. If your data fits in memory then you should use Pandas. **Dask becomes useful when the dataset you want to analyze is larger than your machine's RAM** where you would normally run into `MemoryError`s.\n",
    "\n",
    "```python\n",
    "    MemoryError:  ...\n",
    "```\n",
    "\n",
    "This also means:\n",
    "\n",
    "## Don't use Dask DataFrames if you don't need to!\n",
    "Distributed computing brings a lot of additional complexity into the mix and will **incur overhead**. If your dataset and computations fit comfortably within your local resources **this overhead will may be larger than the performance gain** you'll get by using Dask. In that case, stick with non-distributed libraries like pandas, numpy and scikit-learn. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "389b5226",
   "metadata": {
    "tags": []
   },
   "source": [
    "## About this notebook\n",
    "During this tutorial, we will work with a dataset containg NYC flight data. This dataset is only about 200MB on disk so that you can download it in a reasonable time and exercises finish quickly, but Dask Dataframes will scale to datasets much larger than the memory on your local machine. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cd61890f",
   "metadata": {},
   "source": [
    "## Getting started with Dask DataFrames\n",
    "\n",
    "Let's use Dask DataFrame's to explore the NYC flight dataset. Dask's `read_csv` function supports wildcard characters like `\"*\"` which can be used to load an entire directory of CSV files."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c69dd555",
   "metadata": {},
   "outputs": [],
   "source": [
    "%run prep_data.py -d flights"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5c3b20b3",
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "files = os.path.join('data', 'nycflights', '*.csv')\n",
    "files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "11b23870",
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.dataframe as dd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4a7df915",
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf = dd.read_csv(files,\n",
    "                 parse_dates={'Date': [0, 1, 2]},\n",
    "                 dtype={\"TailNum\": str,\n",
    "                        \"CRSElapsedTime\": float,\n",
    "                        \"Cancelled\": bool})\n",
    "ddf"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c61959ff",
   "metadata": {},
   "source": [
    "Notice that the representation of the dataframe object contains no data - Dask has just done enough to read the start of the first file, and infer the column names and dtypes.\n",
    "\n",
    "**Dask is lazy!**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d714511c",
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5a24944c",
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.dtypes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b4a86ccb",
   "metadata": {},
   "source": [
    "Dask DataFrames have an `.npartitions` attribute which tells you how many partitions make up a Dask DataFrame.\n",
    "\n",
    "Dask is able to process larger-than-memory datasets by cutting computations into smaller parts and processing those in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b04c8168",
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf.npartitions"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "047d2fb2",
   "metadata": {
    "tags": []
   },
   "source": [
    "## The pandas Look & Feel\n",
    "Dask DataFrames implement a well-used portion of the Pandas API in a way that allows for parallel and out-of-core computation. This means that a lot of Dask DataFrame code will look and feel familiar to pandas users: \n",
    "\n",
    "```python\n",
    "import pandas as pd                   import dask.dataframe as dd\n",
    "df = pd.read_csv('2015-01-01.csv')    df = dd.read_csv('2015-*-*.csv')\n",
    "df.groupby(df.user_id).value.mean()   df.groupby(df.user_id).value.mean().compute()\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c87fe45a",
   "metadata": {
    "tags": []
   },
   "source": [
    "This is because, internally, **a Dask DataFrame is composed of many pandas DataFrames**: \n",
    "\n",
    "<img src=\"http://dask.pydata.org/en/latest/_images/dask-dataframe.svg\" width=\"50%\">\n",
    "\n",
    "Dask DataFrames are divided into different **partitions** where each partition is a pandas DataFrame. This is why driving the Dask car *can feel* like you're still driving the pandas car: Dask is performing a bunch of regular pandas operations on regular pandas objects under the hood.\n",
    "\n",
    "But don't forget that you've entered the world of distributed computing now -- which means you've added a lot more complexity to the mix. You now need to consider things like concurrency, state, data duplicates, data loss, etc.\n",
    "\n",
    "Luckily, with a high-level Collection like DataFrames, Dask handles most of these complicated questions for you. "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "67a6d01f",
   "metadata": {
    "tags": []
   },
   "source": [
    "## pandas-like Computations\n",
    "\n",
    "Let's see this in action with a more involved example. Let's compute the largest flight departure delay.\n",
    "\n",
    "In pandas we could do this by iterating over each file to find the individual maximums and then find the final maximum over the individual maximums.\n",
    "\n",
    "```python\n",
    "import pandas as pd\n",
    "\n",
    "files = os.listdir(os.path.join('data', 'nycflights'))\n",
    "\n",
    "maxes = []\n",
    "\n",
    "for file in files:\n",
    "    df = pd.read_csv(os.path.join('data', 'nycflights', file))\n",
    "    maxes.append(df.DepDelay.max())\n",
    "\n",
    "final_max = max(maxes)\n",
    "```\n",
    "\n",
    "Thankfully, we can do this with Dask DataFrames using pandas-like code:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a21dc382",
   "metadata": {},
   "outputs": [],
   "source": [
    "max_delay = ddf[\"DepDelay\"].max()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d3dce768",
   "metadata": {},
   "outputs": [],
   "source": [
    "max_delay"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "83c59405",
   "metadata": {},
   "source": [
    "The above cell looks exactly like what we would do using pandas...but the result does not! \n",
    "\n",
    "Instead of the actual result of the computation, we only get some schematic information. This is because Dask DataFrames are lazily evaluated. This means that **no computation happens unless you explicitly tell Dask to do so** by calling `.compute()`.\n",
    "\n",
    "Before actually performing a computation, dask first constructs a task graph that it can use to optimize computing the result in parallel. You can think of a task graph as the recipe or routemap that contains all the necessary instructions to arrive at the final result. Once you call `.compute()`, Dask will execute the instructions contained in the task graph and perform computations in parallel.\n",
    "\n",
    "Let's look at the task graph to get a feel for how Dask's blocked algorithms work:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e9c5128d",
   "metadata": {},
   "outputs": [],
   "source": [
    "max_delay.visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f0acae36",
   "metadata": {},
   "source": [
    "Some things to note:\n",
    "\n",
    "1.  Up until this point everything is lazy. To evaluate the result for `max_delay`, call its `compute()` method:\n",
    "2.  Dask will delete intermediate results (like the full pandas DataFrame for each file) as soon as possible.\n",
    "    -  This lets us handle datasets that are larger than memory\n",
    "    -  This means that repeated computations will have to load all of the data in each time "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "92bb3578",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time \n",
    "max_delay.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "05540de7",
   "metadata": {
    "tags": []
   },
   "source": [
    "## More Dask DataFrame computations\n",
    "\n",
    "Let's see couple of examples on how the API for Dask DataFrames is the same than Pandas. If you are comfortable with Pandas, the following operations will look very familiar, except we will need to add the `compute()` to get the results wanted.\n",
    "\n",
    "### Example 1: Total of non-cancelled flights taken\n",
    "\n",
    "Notice that there is a column in our DataFrame called `\"Cancelled\"` that is a boolean. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0883849a",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "(~ddf[\"Cancelled\"]).sum().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5df23079",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Example 2: Total of non-cancelled flights taken by airport\n",
    "\n",
    "We should select the non-canceled flights, use the operation `groupby` on the `\"Origin\"` column and finally use `count` to get the detailed per airport."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7b01ecd4",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf[~ddf[\"Cancelled\"]].groupby(\"Origin\")[\"Origin\"].count().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d7dec6b4",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Exercise 1: What is the average departure delay from each airport?"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b5668216",
   "metadata": {},
   "source": [
    "Uncomment and run the cell below to see the solution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6319550d",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf.groupby(\"Origin\")[\"DepDelay\"].mean().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "230c37b2",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Exercise 2: What day of the week has the worst average departure delay?\n",
    "Uncomment and run the cell below to see the solution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4dacd560",
   "metadata": {
    "jupyter": {
     "source_hidden": true
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "ddf.groupby(\"DayOfWeek\")[\"DepDelay\"].mean().idxmax().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b0da1c43",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Working with Partitions\n",
    "Dask DataFrames implements a large portion of the pandas API by simply performing pandas methods on its partitions (which are pandas objects). \n",
    "\n",
    "### Mapping Functions with `map_partitions`\n",
    "However, sometimes you might want to manipulate your Dask DataFrame with a custom function. You can use the `map_partitions` method for this.\n",
    "\n",
    "Imagine you find out that there was a 2-minute error in the `DepDelay` column.\n",
    "\n",
    "Let's create a pandas `apply` function that will subtract 2 from every input:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ca97e0c4",
   "metadata": {},
   "outputs": [],
   "source": [
    "def subtract_2(df):\n",
    "    return df.apply(lambda x: x-2)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "af5ffd7a",
   "metadata": {},
   "source": [
    "We can then map this function over all of our partitions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c5db73a2",
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf[\"Adjusted_DepDelay\"] = ddf[\"DepDelay\"].map_partitions(subtract_2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c6f79e39",
   "metadata": {},
   "outputs": [],
   "source": [
    "ddf[[\"DepDelay\", \"Adjusted_DepDelay\"]].head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "45884f8b",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Performance tip: When to call .compute()\n",
    "\n",
    "In the examples and exercises above, we sometimes perform the same operation more than once (e.g. `read_csv`). Dask DataFrames hashes the arguments, allowing duplicate computations to be shared, and only computed once. You can use `dask.compute()` to merge task graphs of multiple operations.\n",
    "\n",
    "For example, let's compute the mean and standard deviation for departure delay of all non-canceled flights. Since Dask operations are lazy, those values aren't the final results until we `compute` them. They're just the recipe required to get the result.\n",
    "\n",
    "If we compute them with two calls to compute, there is no sharing of intermediate computations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "540952b2",
   "metadata": {},
   "outputs": [],
   "source": [
    "non_cancelled = ddf[~ddf[\"Cancelled\"]]\n",
    "mean_delay = non_cancelled[\"DepDelay\"].mean()\n",
    "std_delay = non_cancelled[\"DepDelay\"].std()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "da852348",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "mean_delay_result = mean_delay.compute()\n",
    "std_delay_result = std_delay.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7ff6c61e",
   "metadata": {},
   "source": [
    "Now, let's see how long it takes if we try computing `mean_delay` and `std_delay` with a single `compute()` call."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1514304e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eefc4c22",
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "mean_delay_res, std_delay_res = dask.compute(mean_delay, std_delay)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "48db683a",
   "metadata": {},
   "source": [
    "Using `dask.compute` takes roughly 1/2 the time. This is because the task graphs for both results are merged when calling `dask.compute`, allowing shared operations (like `read_csv`) to only be done once instead of twice. In particular, using `dask.compute` only does the following once:\n",
    "\n",
    "- The calls to `read_csv`\n",
    "- The filter (`df[~df[\"Cancelled\"]]`)\n",
    "- The `\"DepDelay\"` column indexing\n",
    "- Some of the necessary reductions (`sum`, `count`)\n",
    "\n",
    "To see what the merged task graphs between multiple results look like (and what's shared), you can use the `dask.visualize` function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fc371ed9",
   "metadata": {},
   "outputs": [],
   "source": [
    "dask.visualize(mean_delay, std_delay)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4b983d2a",
   "metadata": {
    "tags": []
   },
   "source": [
    "# Extra resources\n",
    "\n",
    "- Explore applying custom code to Dask DataFrames: [Dask Tutorial DataFrames](https://github.com/dask/dask-tutorial/blob/main/04_dataframe.ipynb)\n",
    "- [Dask DataFrame documentation](https://docs.dask.org/en/latest/dataframe.html)\n",
    "- [Dask DataFrame examples](https://examples.dask.org/dataframe.html)"
   ]
  }
 ],
 "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.9.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}