03-dask-dataframes.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"http://dask.readthedocs.io/en/latest/_images/dask_horizontal.svg\"\n",
    "     align=\"right\"\n",
    "     width=\"30%\"\n",
    "     alt=\"Dask logo\\\">\n",
    "\n",
    "\n",
    "# Dask DataFrames\n",
    "\n",
    "We finished the last section by building a parallel dataframe computation over a directory of CSV files using `dask.delayed`.  In this section we use `dask.dataframe` to build computations for us in the common case of tabular computations.  Dask dataframes look and feel like Pandas dataframes but they run on the same infrastructure that powers `dask.delayed`.\n",
    "\n",
    "In this notebook we use the same airline data as in notebook 01, but now rather than write for loops we let `dask.dataframe` construct our computations for us.  The `dask.dataframe.read_csv` function can take a globstring like `\"data/nycflights/*.csv\"` and build parallel computations on all of our data at once.\n",
    "\n",
    "## When to use `dask.dataframe`\n",
    "\n",
    "Pandas is great for tabular datasets that fit in memory. Dask becomes useful when the dataset you want to analyze is larger than your machine's RAM. We didn't want to overwhelm the conference WiFi downloading large datasets, so the demo dataset we're working with is only about 200MB. But `dask.dataframe` will scale to larger than memory datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "import dask\n",
    "import dask.dataframe as dd\n",
    "import pandas as pd\n",
    "\n",
    "pd.options.display.max_rows = 10\n",
    "\n",
    "df = dd.read_csv(os.path.join('data', 'nycflights', '*.csv'),\n",
    "                 parse_dates={'Date': [0, 1, 2]})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the first 5 rows\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the last 5 rows\n",
    "df.tail()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### What just happened?\n",
    "\n",
    "Unlike `pandas.read_csv` which reads in the entire file before inferring datatypes, `dask.dataframe.read_csv` only reads in a sample from the beginning of the file (or first file if using a glob). These inferred datatypes are then enforced when reading all partitions.\n",
    "\n",
    "In this case, the datatypes inferred in the sample are incorrect. 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). When this happens you have a few options:\n",
    "\n",
    "- Specify dtypes directly using the `dtype` keyword. This is the recommended solution, as it's the least error prone (better to be explicit than implicit) and also the most performant.\n",
    "- Increase the size of the `sample` keyword (in bytes)\n",
    "- Use `assume_missing` to make `dask` assume that columns inferred to be `int` (which don't allow missing values) are actually floats (which do allow missing values). In our particular case this doesn't apply.\n",
    "\n",
    "In our case we'll use the first option and directly specify the `dtypes` of the offending columns. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = dd.read_csv(os.path.join('data', 'nycflights', '*.csv'),\n",
    "                 parse_dates={'Date': [0, 1, 2]},\n",
    "                 dtype={'TailNum': str,\n",
    "                        'CRSElapsedTime': float,\n",
    "                        'Cancelled': bool})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.tail()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Computations with `dask.dataframe`\n",
    "\n",
    "We compute the maximum of the `DepDelay` column. With just pandas, we would loop over each file to find the individual maximums, then find the final maximum over all the individual maximums\n",
    "\n",
    "```python\n",
    "maxes = []\n",
    "for fn in filenames:\n",
    "    df = pd.read_csv(fn)\n",
    "    maxes.append(df.DepDelay.max())\n",
    "    \n",
    "final_max = max(maxes)\n",
    "```\n",
    "\n",
    "We could wrap that `pd.read_csv` with `dask.delayed` so that it runs in parallel. Regardless, we're still having to think about loops, intermediate results (one per file) and the final reduction (`max` of the intermediate maxes). This is just noise around the real task, which pandas solves with\n",
    "\n",
    "```python\n",
    "df = pd.read_csv(filename, dtype=dtype)\n",
    "df.DepDelay.max()\n",
    "```\n",
    "\n",
    "`dask.dataframe` lets us write pandas-like code, that operates on larger than memory datasets in parallel."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time df.DepDelay.max().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This writes the delayed computation for us and then runs it.  \n",
    "\n",
    "Some things to note:\n",
    "\n",
    "1.  As with `dask.delayed`, we need to call `.compute()` when we're done.  Up until this point everything is lazy.\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 (run the code above again, is it faster or slower than you would expect?)\n",
    "    \n",
    "As with `Delayed` objects, you can view the underlying task graph using the `.visualize` method:/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.DepDelay.max().visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercises\n",
    "\n",
    "In this section we do a few `dask.dataframe` computations. If you are comfortable with Pandas then these should be familiar. You will have to think about when to call `compute`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.) How many rows are in our dataset?\n",
    "\n",
    "If you aren't familiar with pandas, how would you check how many records are in a list of tuples?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here...\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/03-dask-dataframe-rows.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.) In total, how many non-canceled flights were taken?\n",
    "\n",
    "With pandas, you would use [boolean indexing](https://pandas.pydata.org/pandas-docs/stable/indexing.html#boolean-indexing)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here...\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/03-dask-dataframe-non-cancelled.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.) In total, how many non-cancelled flights were taken from each airport?\n",
    "\n",
    "*Hint*: use [`df.groupby`](https://pandas.pydata.org/pandas-docs/stable/groupby.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here...\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/03-dask-dataframe-non-cancelled-per-airport.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.) What was the average departure delay from each airport?\n",
    "\n",
    "Note, this is the same computation you did in the previous notebook (is this approach faster or slower?)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here...\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/03-dask-dataframe-delay-per-airport.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5.) What day of the week has the worst average departure delay?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here...\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/03-dask-dataframe-delay-per-day.py"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sharing Intermediate Results\n",
    "\n",
    "When computing all of the above, we sometimes did the same operation more than once. For most operations, `dask.dataframe` hashes the arguments, allowing duplicate computations to be shared, and only computed once.\n",
    "\n",
    "For example, lets compute the mean and standard deviation for departure delay of all non-canceled flights:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "non_cancelled = df[~df.Cancelled]\n",
    "mean_delay = non_cancelled.DepDelay.mean()\n",
    "std_delay = non_cancelled.DepDelay.std()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since dask operations are lazy, those values aren't the final results yet. They're just the recipe require to get the result."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "non_cancelled"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Using two calls to `.compute`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "mean_delay_res = mean_delay.compute()\n",
    "std_delay_res = std_delay.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Using one call to `dask.compute`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "mean_delay_res, std_delay_res = dask.compute(mean_delay, std_delay)"
   ]
  },
  {
   "cell_type": "markdown",
   "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 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",
    "- 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 (we might want to use `filename='graph.pdf'` to zoom in on the graph better):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dask.visualize(mean_delay, std_delay)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dask DataFrame Data Model\n",
    "\n",
    "For the most part, a Dask DataFrame feels like a pandas DataFrame.\n",
    "So far, the biggest difference we've seen is that Dask operations are lazy; they build up a task graph instead of executing immediately (more details coming in [Schedulers](04-schedulers.ipynb)).\n",
    "This lets Dask do operations in parallel and out of core.\n",
    "\n",
    "In [Dask Arrays](02-dask-arrays.ipynb), we saw that a `dask.array` was composed of many NumPy arrays, chunked along one or more dimensions.\n",
    "It's similar for `dask.dataframe`: a Dask DataFrame is composed of many pandas DataFrames. For `dask.dataframe` the chunking happens only along the index.\n",
    "\n",
    "<img src=\"http://dask.pydata.org/en/latest/_images/dask-dataframe.svg\" width=\"30%\">\n",
    "\n",
    "We call each chunk a *partition*, and the upper / lower bounds are *divisions*.\n",
    "Dask *can* store information about the divisions. We'll cover this in more detail in [Distributed DataFrames](05-distributed-dataframes-and-efficiency.ipynb).\n",
    "For now, partitions come up when you write custom functions to apply to Dask DataFrames"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Converting `CRSDepTime` to a timestamp\n",
    "\n",
    "This dataset stores timestamps as `HHMM`, which are read in as integers in `read_csv`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "crs_dep_time = df.CRSDepTime.head(10)\n",
    "crs_dep_time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To convert these to timestamps of scheduled departure time, we need to convert these integers into `pd.Timedelta` objects, and then combine them with the `Date` column.\n",
    "\n",
    "In pandas we'd do this using the `pd.to_timedelta` function, and a bit of arithmetic:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "# Get the first 10 dates to complement our `crs_dep_time`\n",
    "date = df.Date.head(10)\n",
    "\n",
    "# Get hours as an integer, convert to a timedelta\n",
    "hours = crs_dep_time // 100\n",
    "hours_timedelta = pd.to_timedelta(hours, unit='h')\n",
    "\n",
    "# Get minutes as an integer, convert to a timedelta\n",
    "minutes = crs_dep_time % 100\n",
    "minutes_timedelta = pd.to_timedelta(minutes, unit='m')\n",
    "\n",
    "# Apply the timedeltas to offset the dates by the departure time\n",
    "departure_timestamp = date + hours_timedelta + minutes_timedelta\n",
    "departure_timestamp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.UniqueCarrier.str.slice(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Custom code and Dask Dataframe\n",
    "\n",
    "We could swap out `pd.to_timedelta` for `dd.to_timedelta` and do the same operations on the entire dask DataFrame. But let's say that Dask hadn't implemented a `dd.to_timedelta` that works on Dask DataFrames. What would you do then?\n",
    "\n",
    "`dask.dataframe` provides a few methods to make applying custom functions to Dask DataFrames easier:\n",
    "\n",
    "- [`map_partitions`](http://dask.pydata.org/en/latest/dataframe-api.html#dask.dataframe.DataFrame.map_partitions)\n",
    "- [`map_overlap`](http://dask.pydata.org/en/latest/dataframe-api.html#dask.dataframe.DataFrame.map_overlap)\n",
    "- [`reduction`](http://dask.pydata.org/en/latest/dataframe-api.html#dask.dataframe.DataFrame.reduction)\n",
    "\n",
    "Here we'll just be discussing `map_partitions`, which we can use to implement `to_timedelta` on our own:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Look at the docs for `map_partitions`\n",
    "\n",
    "help(df.CRSDepTime.map_partitions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The basic idea is to apply a function that operates on a DataFrame to each partition.\n",
    "In this case, we'll apply `pd.to_timedelta`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hours = df.CRSDepTime // 100\n",
    "# hours_timedelta = pd.to_timedelta(hours, unit='h')\n",
    "hours_timedelta = hours.map_partitions(pd.to_timedelta, unit='h')\n",
    "\n",
    "minutes = df.CRSDepTime % 100\n",
    "# minutes_timedelta = pd.to_timedelta(minutes, unit='m')\n",
    "minutes_timedelta = minutes.map_partitions(pd.to_timedelta, unit='m')\n",
    "\n",
    "departure_timestamp = df.Date + hours_timedelta + minutes_timedelta"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "departure_timestamp"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "departure_timestamp.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise: Rewrite above to use a single call to `map_partitions`\n",
    "\n",
    "This will be slightly more efficient than two separate calls, as it reduces the number of tasks in the graph."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def compute_departure_timestamp(df):\n",
    "    # TODO"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "departure_timestamp = df.map_partitions(compute_departure_timestamp)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "departure_timestamp.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/03-dask-dataframe-map-partitions.py"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}