05-distributed-dataframes-and-efficiency.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",
    "# Personal Dask Cluster\n",
    "\n",
    "Go to http://<TBD>/hub/login for your own Dask cluster. Use your first and last name provided to SciPy for the username and \"dask\" for the password.\n",
    "(We don't actually do any authentication).\n",
    "\n",
    "# Distributed DataFrames and Efficiency\n",
    "\n",
    "In the previous notebooks we discussed `dask.dataframe` and `dask.distributed`. Here we combine them on a larger dataset, and discuss efficiency and performance tips.\n",
    "\n",
    "We will cover the following topics:\n",
    "\n",
    "1. Persist common intermediate results in memory with `persist`\n",
    "2. Partitions and partition size\n",
    "3. Using indices to improve efficiency"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from scipy_utils import make_cluster\n",
    "from dask.distributed import Client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The next cell will start up some workers for you. This make take a few minutes, but they widget will update automatically when the workers are ready. You don't need to do anything with the manual or adaptive scaling."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster = make_cluster()\n",
    "cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Be sure to open the diagnostics UI.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Moving to distributed\n",
    "\n",
    "A few things change when moving from local to distributed computing.\n",
    "\n",
    "1. Environment: Each worker is a separate machine, and needs to have the required libraries installed. This cluster was setup using [Kubernetes](http://dask.pydata.org/en/latest/setup/kubernetes.html#).\n",
    "2. File system: Previously, every worker (threads, processes, or even the distributed scheduler in local mode) had access to your laptops file system. In a distributed environment, you'll need some kind of shared file system to read data (cloud storage like S3 or GCS, or a network file system)\n",
    "3. Communication: Moving data between machines is relatively expensive. When possible, the distributed scheduler will ensure that tasks are scheduled to be run on workers that already have the required data. But some tasks will require data from multiple machines."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## The full airline dataset\n",
    "\n",
    "We have the full airline dataset stored on `GCS`. This is the same as the one you've been working with, but includes all originating airports and a few extra columns. We change the `read_csv` call slightly to avoid the extra columns.\n",
    "\n",
    "`dask.dataframe` has support for reading directly from `GCS`, so we can use our `read_csv` call from before."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.dataframe as dd\n",
    "\n",
    "columns = ['Year', 'Month', 'DayofMonth', 'DayOfWeek', 'DepTime', 'CRSDepTime',\n",
    "           'ArrTime', 'CRSArrTime', 'UniqueCarrier', 'FlightNum', 'TailNum',\n",
    "           'ActualElapsedTime', 'CRSElapsedTime', 'AirTime', 'ArrDelay',\n",
    "           'DepDelay', 'Origin', 'Dest', 'Distance', 'TaxiIn', 'TaxiOut',\n",
    "           'Cancelled']\n",
    "\n",
    "df = dd.read_csv('gcs://anaconda-public-data/airline/(199)|(200)*.csv',\n",
    "                 parse_dates={'Date': [0, 1, 2]},\n",
    "                 dtype={'TailNum': object,\n",
    "                        'CRSElapsedTime': float,\n",
    "                        'Distance': float,\n",
    "                        'Cancelled': bool},\n",
    "                 usecols=columns)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Persist data in distributed memory\n",
    "\n",
    "Every time we run an operation like `df[~df.Cancelled].DepDelay.max().compute()` we read through our dataset from disk.  This can be slow, especially because we're reading data from CSV.  We usually have two options to make this faster:\n",
    "\n",
    "1.  Persist relevant data in memory, either on our computer or on a cluster\n",
    "2.  Use a faster on-disk format, like HDF5 or Parquet\n",
    "\n",
    "In this section we persist our data in memory.  On a single machine this is often done by doing a bit of pre-processing and data reduction with dask dataframe and then `compute`-ing to a Pandas dataframe and using Pandas in the future.  \n",
    "\n",
    "```python\n",
    "df = dd.read_csv(...)\n",
    "df = df[df.Origin == 'LGA']  # filter down to smaller dataset\n",
    "pdf = df.compute()  # convert to pandas\n",
    "pdf ... # continue with familiar Pandas workflows\n",
    "```\n",
    "\n",
    "However on a distributed cluster when even our cleaned data is too large we still can't use Pandas.  In this case we ask Dask to persist data in memory with the `dask.persist` function.  This is what we'll do today.  This will help us to understand when data is lazy and when it is computing.\n",
    "\n",
    "You can trigger computations using the persist method:\n",
    "\n",
    "    x = x.persist()\n",
    "\n",
    "or the dask.persist function for multiple inputs:\n",
    "\n",
    "    x, y = dask.persist(x, y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise\n",
    "\n",
    "Persist the dataframe into memory.\n",
    "\n",
    "- How long does the cell take to execute (look at the \"busy\" indicator in the top-right)?\n",
    "- After it has persisted how long does it take to compute `df[~df.Cancelled].DepDelay.count().compute()`?\n",
    "- Looking at the plots in the diagnostic web page (the link was printed above), what is taking up most of the time? (You can over over rectangles to see what function they represent)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = # TODO: persist dataframe in memory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time _ = df.Cancelled[~df.Cancelled].count().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise\n",
    "\n",
    "Repeat the groupby computation from the previous notebooks. What is taking all of the time now?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# What was the average departure delay from each airport?\n",
    "df[~df.Cancelled].groupby('Origin').DepDelay.mean().nlargest(10).compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Partitions\n",
    "\n",
    "One `dask.dataframe` is composed of several Pandas dataframes.  The organization of these dataframes can significantly impact performance.  In this section we discuss two common factors that commonly impact performance:\n",
    "\n",
    "1. The number of Pandas dataframes can affect overhead.  If the dataframes are too small then Dask might spend more time deciding what to do than Pandas spends actually doing it.  Ideally computations should take 100's of milliseconds.\n",
    "\n",
    "2. If we know how the dataframes are sorted then certain operations become much faster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Number of partitions and partition size\n",
    "\n",
    "When we read in our data from CSV files we get potentially multiple Pandas dataframe for each file. Look at the metadata below to determine a few things about the current partitioning:\n",
    "- How many partitions are there?\n",
    "- Are the splits along the index between partitions known? If so, what are they?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Number of partitions\n",
    "df.npartitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Are the splits between partitions known?\n",
    "df.known_divisions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The splits between partitions. If unknown these are all `None`\n",
    "df.divisions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise: How large is the DataFrame?\n",
    "\n",
    "- How would you compute the memory usage of a single pandas DataFrame?\n",
    "- Given your knowledge of Dask, how would you do it for a Dask DataFrame?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Your code here...\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load solutions/05-distributed-dataframes-memory-usage.ipynb"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sorted Index column\n",
    "\n",
    "*This section doesn't have any exercises.  Just follow along.*\n",
    "\n",
    "Many dataframe operations like loc-indexing, groupby-apply, and joins are *much* faster on a sorted index.  For example, if we want to get data for a particular day of data it *really* helps to know where that day is, otherwise we need to search over all of our data.\n",
    "\n",
    "The Pandas model gives us a sorted index column.  Dask.dataframe copies this model, and it remembers the min and max values of every partition's index.\n",
    "\n",
    "By default, our data doesn't have an index."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So if we search for a particular day it takes a while because it has to pass through all of the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time df[df.Date == '1992-05-05'].compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df[df.Date == '1992-05-05'].visualize(optimize_graph=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "However if we set the `Date` column as the index then this operation can be much much faster.\n",
    "\n",
    "Calling `set_index` followed by `persist` results in a new set of dataframe partitions stored in memory, sorted along the index column. To do this dask has to\n",
    "\n",
    "- Shuffle the data by date, resulting in the same number of output partitions\n",
    "- Set the index for each partition\n",
    "- Store the resulting partitions in distributed memory\n",
    "\n",
    "This can be a (relatively) expensive operation, but allows certain queries to be more optimized. \n",
    "\n",
    "Watch the diagnostics page while the next line is running to see how the shuffle and index operation progresses."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "df = df.set_index('Date').persist()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After the index is set, we now have known divisions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Number of partitions\n",
    "df.npartitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Are the splits between partitions known?\n",
    "df.known_divisions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The splits between partitions.\n",
    "df.divisions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The repr for a dask dataframe can also be useful for\n",
    "# seeing partition information\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Repeating the same query for all flights on a specific date, we can see that we're much faster after setting the index:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%time df.loc['1992-05-05'].compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you look at the resulting graph, you can see that dask was able to optimize the computation to only look at a single partition:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.loc['1992-05-05'].visualize(optimize_graph=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Timeseries operations\n",
    "\n",
    "When the index of a dask dataframe is a known `DatetimeIndes`, traditional pandas timeseries operations are supported. For example, now that we have a sorted index we can resample the `DepDelay` column into 1 month bins."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time \n",
    "(df.DepDelay\n",
    "   .resample('1M')\n",
    "   .mean()\n",
    "   .fillna(method='ffill')\n",
    "   .compute()\n",
    "   .plot(figsize=(10, 5)));"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# When you're done with the `airlines` dataset\n",
    "client.restart()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise: Explore the NYC Taxi dataset\n",
    "\n",
    "We have some of the NYC Taxi ride dataset in parquet format stored in GCS."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "taxi = dd.read_parquet(\"gcs://anaconda-public-data/nyc-taxi/nyc.parquet\")\n",
    "taxi.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Some questions?\n",
    "\n",
    "- How large is the dataset? Will it fit in your cluster's RAM if you persist it?\n",
    "- What's the average tip percent by hour?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# clean up, when finished with the notebook\n",
    "client.close()\n",
    "cluster.close()"
   ]
  }
 ],
 "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
}