01-dataframes.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "Dask Dataframes on NYC Taxi Data\n",
    "================================\n",
    "\n",
    "<img src=\"http://pandas.pydata.org/_static/pandas_logo.png\"\n",
    "     align=\"left\"\n",
    "     width=\"30%\"\n",
    "     alt=\"Pandas logo\">\n",
    "     <img src=\"http://dask.readthedocs.io/en/latest/_images/dask_horizontal.svg\"\n",
    "     align=\"right\"\n",
    "     width=\"30%\"\n",
    "     alt=\"Dask logo\">\n",
    "\n",
    "In this section we will learn how to ...\n",
    "\n",
    "-  use Dask Dataframe to scale Pandas workloads\n",
    "-  call `.compute` and `.persist` to trigger computation\n",
    "-  start and scale a Dask cluster on Kubernetes\n",
    "-  interpret dashboard plots\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import warnings\n",
    "\n",
    "warnings.filterwarnings(\"ignore\", message=\"numpy.dtype size changed\")\n",
    "warnings.filterwarnings(\"ignore\", message=\"numpy.ufunc size changed\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## We have several CSV files in cloud storage"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "from gcsfs import GCSFileSystem\n",
    "gcs = GCSFileSystem()\n",
    "\n",
    "sorted(gcs.glob('anaconda-public-data/nyc-taxi/csv/2015/yellow_*.csv'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Read a subset with Pandas\n",
    "\n",
    "It's too big to fit in memory on a single machine, so we pull out the first million rows to get a first impression."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "with gcs.open('anaconda-public-data/nyc-taxi/csv/2015/yellow_tripdata_2015-01.csv') as f:\n",
    "    df = pd.read_csv(f, nrows=1000000, parse_dates=['tpep_pickup_datetime', 'tpep_dropoff_datetime'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Investigate the subset as normal"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# How many passengers total?\n",
    "df.passenger_count.sum()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The average trip distance for rides with a single passenger\n",
    "df2 = df[df.passenger_count == 1]\n",
    "df2.trip_distance.mean()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The average trip distance grouped by passenger counts\n",
    "df.groupby(df.passenger_count).trip_distance.mean()  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Start a Dask Cluster\n",
    "\n",
    "Your notebook is conveniently attached to a Kubernetes cluster, so you can start a Dask cluster using the [dask-kubernetes](https://kubernetes.dask.org/en/latest/) project.\n",
    "\n",
    "For more information on deploying Dask on different cluster technology see [Dask's deployment documentation](https://docs.dask.org/en/latest/setup.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask_kubernetes import KubeCluster\n",
    "cluster = KubeCluster(n_workers=10)\n",
    "cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "\n",
    "client = Client(cluster)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create Dask dataframe around all of the data\n",
    "\n",
    "Before we loaded only a subset of one CSV file.  Now lets use Dask dataframe to read all of the files.\n",
    "\n",
    "For more information you can read [Dask's documentation for creating dataframes](http://docs.dask.org/en/latest/dataframe-create.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.dataframe as dd\n",
    "\n",
    "df = dd.read_csv('gcs://anaconda-public-data/nyc-taxi/csv/2015/yellow_*.csv', \n",
    "                 parse_dates=['tpep_pickup_datetime', 'tpep_dropoff_datetime'])\n",
    "df = df.persist()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask dataframes look like Pandas dataframes, and support most of the common Pandas methods."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.passenger_count.sum()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Investigate laziness and use the `.compute()` method\n",
    "\n",
    "Note that the `df.passenger_count.sum()` computation did not yet execute.  Dask dataframes are *lazy* by default, so they only evaluate when we tell them to.\n",
    "\n",
    "There are two ways to trigger computation:\n",
    "\n",
    "-  `result = result.compute()`: triggers computation and stores the result into local memory as a Pandas object.  \n",
    "\n",
    "    You should use this with *small* results that will fit into memory.\n",
    "-  `result = result.persist()`: triggers computation and stores the result into distributed memory, returning another Dask dataframe object.  \n",
    "\n",
    "    You should use this with *large* results that you want to stage in distributed memory for repeated computation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### *Exercise*: Run the Pandas computations above with Dask dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# How many passengers total?\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The average trip distance for rides with a single passenger\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The average trip distance grouped by passenger counts\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### *Question*: When is it safe to call compute?\n",
    "\n",
    "Recall that calling `.compute()` on a Dask DataFrame returns a Pandas result in your local memory.  This can be dangerous if the size of the result is large.  In which of the following situations is calling `.compute()` ok?\n",
    "\n",
    "-  `df.sum()`\n",
    "-  `df[df.passenger_count == 1]`\n",
    "-  `df[df.passenger_count == 10]`\n",
    "-  `df.groupby(df.passenger_count).trip_distance.mean()`\n",
    "-  `df.groupby(df.tpep_pickup_latitute).trip_distance.mean()`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Persist data in memory\n",
    "\n",
    "When we started this notebook we ran the following lines to create our dataframe.\n",
    "\n",
    "```python\n",
    "df = dd.read_csv('gcs://anaconda-public-data/nyc-taxi/csv/2015/yellow_*.csv', \n",
    "                 parse_dates=['tpep_pickup_datetime', 'tpep_dropoff_datetime'])\n",
    "df = df.persist()\n",
    "```\n",
    "\n",
    "In particular, we called `df = df.persist()` to load all of the CSV data into distributed memory.  Having this data in memory made our subsequent computations fast.  \n",
    "\n",
    "In this section we're going to reset our cluster and run the same computations, but without persisting our data in memory.  What happens to our computation times?  Why?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.restart()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = dd.read_csv('gcs://anaconda-public-data/nyc-taxi/csv/2015/yellow_*.csv', \n",
    "                 parse_dates=['tpep_pickup_datetime', 'tpep_dropoff_datetime'])\n",
    "df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# How many passengers total?\n",
    "df.passenger_count.sum().compute()  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The average trip distance for rides with a single passenger\n",
    "df2 = df[df.passenger_count == 1]  \n",
    "df2.trip_distance.mean().compute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# The average trip distance grouped by passenger counts\n",
    "df.groupby(df.passenger_count).trip_distance.mean().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### *Exercise*: What did our workers spend their time doing?\n",
    "\n",
    "To answer this question look at the Task Stream dashboard plot.  It will tell you the activity on each core of your cluster (y-axis) over time (x-axis).  You can hover over each rectangle of this plot to determine what kind of task it was.  What kinds of tasks are most common and take up the most time?\n",
    "\n",
    "*Extra*: if you're ahead of the group you might also want to look at the Profile dashboard plot.  You can access this by selecting the orange Dask icon on the left side of your JupyterLab page.  The profile plot is an interactive [Flame graph](http://www.brendangregg.com/FlameGraphs/cpuflamegraphs.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.persist()  # we persist our data again, just to make future sections faster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dask DataFrame Design\n",
    "\n",
    "We briefly discuss the design of Dask dataframes.  Then we follow this section with exercises that dive into this design.\n",
    "\n",
    "<img src=\"http://docs.dask.org/en/latest/_images/dask-dataframe.svg\"\n",
    "     width=\"50%\">\n",
    "     \n",
    "Dask dataframes are composed of many *partitions*, split along the index.  Each partition is a Pandas dataframe or Series.  You can see the number of partitions in the rendering of a Dask Dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And the type of each partition using the `map_partitions` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.map_partitions(type).compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Divisions and the Index\n",
    "\n",
    "Just like Pandas, Dask Dataframe has an *index*, a special column that indexes the rows of our dataframe.  In Dask this index has an additional purpose, it serves as a sorted partitioning of our data.  This makes some algorithms more efficient.  In this section, we'll sort our data by time and dive into the index a bit more deeply.\n",
    "\n",
    "First, notice that our index is not particularly informative.  This is common when you load a dataset from CSV data, which generally doesn't store index or sorting information.\n",
    "\n",
    "Lets set a new index to be the pickup time.  Sorting in parallel is hard, so this is an expensive operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2 = df.set_index('tpep_pickup_datetime').persist()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2.tail()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our dataframe is split into roughly as many partitions as before, but now we know the time range of each partition.  Internally, the divisions between partitions is stored in the divisions attribute."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2.divisions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Question: What took up the most time in the operation above?\n",
    "\n",
    "What colors are most prominent in the task stream plot?\n",
    "\n",
    "When you hover over some of these bars, what do they say?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fast operations along the index\n",
    "\n",
    "Having a sorted dataframe allows for fast operations, like random access lookup and timeseries operations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2.loc['2015-05-05'].compute()  # pick out one day of data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df2.passenger_count.resample('1h').mean().compute().plot()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercises if you are done early\n",
    "\n",
    "Explore timeseries operations like `resample` and `rolling`\n",
    "\n",
    "You may want to look at the [DataFrame API](http://docs.dask.org/en/latest/dataframe-api.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Close things when you're done\n",
    "\n",
    "Before you move onto the next notebook, please close down your current cluster.\n",
    "\n",
    "Alternatively, you can restart this notebook by pressing the `\"0\"` key twice"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "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.7.1"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}