07_dataframe.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"images/dask_horizontal.svg\" align=\"right\" width=\"30%\">"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Table of Contents\n",
    "* [DataFrame](#DataFrame)\n",
    "\t* [Setup](#Setup)\n",
    "\t* [Load Data from CSVs and inspect the dask graph](#Load-Data-from-CSVs-and-inspect-the-dask-graph)\n",
    "\t\t* [`dask.dataframe.read_csv`](#dask.dataframe.read_csv)\n",
    "\t\t* [Exercise: Inspect dask graph](#Exercise:-Inspect-dask-graph)\n",
    "\t* [How does this compare to Pandas?](#How-does-this-compare-to-Pandas?)\n",
    "\t\t* [Features and Size](#Features-and-Size)\n",
    "\t\t* [Speed](#Speed)\n",
    "\t\t* [Exercise: Recall and use Pandas API](#Exercise:-Recall-and-use-Pandas-API)\n",
    "\t\t* [Exercise](#Exercise)\n",
    "\t* [Limitations](#Limitations)\n",
    "\t\t* [What doesn't work?](#What-doesn't-work?)\n",
    "\t\t* [What definitely works?](#What-definitely-works?)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# be sure to shut down other kernels running distributed clients\n",
    "from dask.distributed import Client\n",
    "client = Client()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# DataFrame"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"images/pandas_logo.png\" align=\"right\" width=\"40%\">\n",
    "In this section we use `dask.dataframe` to automatically build and execute dask graphs which process large volumes of CSV files automatically.\n",
    "\n",
    "The `dask.dataframe` module implements a blocked parallel `DataFrame` object that mimics a large subset of the Pandas `DataFrame`. One dask `DataFrame` is comprised of many in-memory pandas `DataFrames` separated along the index. One operation on a dask `DataFrame` triggers many pandas operations on the constituent pandas `DataFrame`s in a way that is mindful of potential parallelism and memory constraints.\n",
    "\n",
    "**Related Documentation**\n",
    "\n",
    "*  [Dask DataFrame documentation](http://dask.pydata.org/en/latest/dataframe.html)\n",
    "*  [Pandas documentation](http://pandas.pydata.org/)\n",
    "\n",
    "**Main Take-aways**\n",
    "\n",
    "1.  Dask.dataframe should be familiar to Pandas users\n",
    "2.  The index grows to include partitions, which are important for efficient queries"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setup"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We create artifical data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from prep import accounts_csvs\n",
    "accounts_csvs(3, 1000000, 500)\n",
    "\n",
    "import os\n",
    "filename = os.path.join('data', 'accounts.*.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load Data from CSVs and inspect the dask graph"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this section we'll use `dask.dataframe` to read in many CSV files at once and compute their total length, using a more Pandas-like interface rather than by playing with iterables and dictionaries manually."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### `dask.dataframe.read_csv`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This works just like `pandas.read_csv`, except on multiple csv files at once."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "filename"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import dask.dataframe as dd\n",
    "df = dd.read_csv(filename)\n",
    "# load and count number of rows\n",
    "%time len(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "What happened here?\n",
    "- Dask investigated the input path and found that there are three matching files \n",
    "- a set of jobs was intelligently created for each chunk - one per original CSV file in this case\n",
    "- each file was loaded into a pandas dataframe, had `len()` applied to it\n",
    "- the subtotals were combined to give you the final grant total."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise: Inspect dask graph"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask `DataFrame` copies a subset of the Pandas API.  \n",
    "\n",
    "However unlike Pandas, operations on dask.dataframes don't trigger immediate computation, instead they add key-value pairs to an underlying dask graph. Recall that in the diagram below, circles are operations and rectangles are results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df.amount.sum().visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Above we see graphs corresponding to a call to `dd.read_csv` and `df.amount.sum()` on the result.  \n",
    "\n",
    "Try changing around the expression `df.amount.sum()` and see how the graph changes.  Explore a bit with the Pandas syntax that you already know."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# try changing the expression and visualize the generated graph\n",
    "df.visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How does this compare to Pandas?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Features and Size"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Pandas is more mature and fully featured than `dask.dataframe`.  If your data fits in memory then you should use Pandas.  The `dask.dataframe` module gives you a limited `pandas` experience when you operate on datasets that don't fit comfortably in memory.\n",
    "\n",
    "During this tutorial we provide a small dataset consisting of a few CSV files.  This dataset is 45MB on disk that expands to about 400MB in memory (the difference is caused by using `object` dtype for strings).  This dataset is small enough that you would normally use Pandas.\n",
    "\n",
    "We've chosen this size so that exercises finish quickly.  Dask.dataframe only really becomes meaningful for problems significantly larger than this, when Pandas breaks with the dreaded \n",
    "\n",
    "    MemoryError:  ...\n",
    "    \n",
    "Furthermore, the distributed scheduler allows the same dataframe expressions to be executed across a cluster. To enable massive \"big data\" processing, one could execute data ingestion functions such as `read_csv`, where the data is held on storage accessible to every worker node (e.g., amazon's S3), and because most operations begin by selecting only some columns, transforming and filtering the data, only relatively small amounts of data need to be communicated between the machines."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Speed"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask.dataframe operations use `pandas` operations internally.  Generally they run at about the same speed except in the following two cases:\n",
    "\n",
    "1.  Dask introduces a bit of overhead, around 1ms per task.  This is usually negligible.\n",
    "2.  When Pandas releases the GIL (coming to `groupby` in the next version) `dask.dataframe` can call several pandas operations in parallel within a process, increasing speed somewhat proportional to the number of cores. For operations which don't release the GIL, multiple processes would be needed to get the same speedup."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise: Recall and use Pandas API"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you are already familiar with the Pandas API then you should have a firm grasp on how to use `dask.dataframe`.  There are a couple of small changes.\n",
    "\n",
    "As noted above, computations on dask `DataFrame` objects don't perform work, instead they build up a dask graph.  We can evaluate this dask graph at any time using the `.compute()` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "result = df.amount.mean()  # create lazily evaluated result\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "result.compute()           # perform actual computation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Aside:** Calling `compute()` caused the execution of the whole graph, returned the value, and then freed up the associated resources. This is normal execution, which leaved all memory available for whatever is next. If we had called `client.compute(result)`, this would have given us a *future*, and instantly returned - the only advantage being that we can do other work while the cluster performs its actions, and we keep a handle to the result (which is cached while the future lives). In order to avoid going through the CSV load every time, we can *persist* the loaded dataframe into the cluster memory - if the cluster has enough memory. This is a very typical step for load-once, query-often data:\n",
    "```python\n",
    "df = client.persist(df)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Try the following exercises\n",
    "\n",
    "1.  Use the `head()` method to get the first ten rows\n",
    "2.  Use the `drop_duplicates()` method to find all of the distinct names\n",
    "3.  Use selections `df[...]` to find how many positive and negative amounts there are\n",
    "4.  Use groupby `df.groupby(df.A).B.func()` to get the average amount per user ID\n",
    "5.  Sort the result to (4) by amount, find the names of the top 10 \n",
    "\n",
    "This section should be easy if you are familiar with Pandas.  If you aren't then that's ok too.  You may find the [pandas documenation](http://pandas.pydata.org/) a useful read in the future.  Don't worry, future sections in this tutorial will not depend on this knowledge."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 1. Use the `head()` method to get the first ten rows\n",
    "#    Note, head computes by default, this is the only operation that doesn't need an explicit call to .compute()\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 2. Use the `drop_duplicates()` method to find all of the distinct names\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 3. Use selections `df[...]` to find how many positive and negative amounts there are\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 3. Use selections `df[...]` to find how many positive and negative amounts there are\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 4. Use groupby `df.groupby(df.A).B.func()` to get the average amount per user ID \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# 5. Combine your answers to 3 and 4 to compute the average withdrawal (negative amount) per name\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# %load solutions/DataFrame-01.py\n",
    "# 1. Use the `head()` method to get the first ten rows\n",
    "df.head()\n",
    "\n",
    "# 2. Use the `drop_duplicates()` method to find all of the distinct names\n",
    "df.names.drop_duplicates().compute()\n",
    "\n",
    "# 3. Use selections `df[...]` to find how many positive and negative amounts\n",
    "# there are\n",
    "len(df[df.amount < 0])\n",
    "\n",
    "# 3. Use selections `df[...]` to find how many positive and negative amounts\n",
    "# there are\n",
    "len(df[df.amount > 0])\n",
    "\n",
    "# 4. Use groupby `df.groupby(df.A).B.func()` to get the average amount per user\n",
    "# ID\n",
    "df.groupby(df.names).amount.mean().compute()\n",
    "\n",
    "# 5. Combine your answers to 3 and 4 to compute the average withdrawal\n",
    "# (negative amount) per name\n",
    "df2 = df[df.amount < 0]\n",
    "df2.groupby(df2.names).amount.mean().compute()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"images/dask-dataframe.svg\" align=\"right\" width=\"40%\">\n",
    "\n",
    "Divisions and the Index\n",
    "---------------------------\n",
    "\n",
    "The Pandas index associates a value to each record/row of your data.  Operations that align with the index, like `loc` can be a bit faster as a result.\n",
    "\n",
    "In `dask.dataframe` this index becomes even more important.  Recall that one dask `DataFrame` consists of several Pandas `DataFrame`s.  These dataframes are separated along the index by value.  For example, when working with time series we may partition our large dataset by month.\n",
    "\n",
    "Recall that these many partitions of our data may not all live in memory at the same time, instead they might live on disk; we simply have tasks that can materialize these pandas `DataFrames` on demand.\n",
    "\n",
    "Partitioning your data can greatly improve efficiency.  Operations like `loc`, `groupby`, and `merge/join` along the index are *much more efficient* than operations along other columns.  You can see how your dataset is partitioned with the `.divisions` attribute.  Note that data that comes out of simple data sources like CSV files aren't intelligently indexed by default.  In these cases the values for `.divisions` will be `None.`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df = dd.read_csv(filename)\n",
    "df.divisions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "However if we set the index to some new column then dask will divide our data roughly evenly along that column and create new divisions for us.  Warning, `set_index` triggers immediate computation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df2 = df.set_index('names')\n",
    "df2.divisions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We see here the minimum and maximum values (\"Alice\" and \"Zelda\") as well as two intermediate values that separate our data well.  This dataset has three partitions.\n",
    "\n",
    "**Aside:** immediately after expensive operations such as set_index would also be a good time to use `client.persist` if the cluster has enough memory."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df2.npartitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df2.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Operations like `loc` only need to load the relevant partitions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df2.loc['Edith']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "df2.loc['Edith'].compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Exercise"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Make a new dataframe that sets the index to the `id` column.  Use `loc` to collect the records with the 100th id."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# %load solutions/DataFrame-02.py\n",
    "df2 = df.set_index('id')\n",
    "print(df2.loc[100])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Limitations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### What doesn't work?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask.dataframe only covers a small but well-used portion of the Pandas API.\n",
    "This limitation is for two reasons:\n",
    "\n",
    "1.  The Pandas API is *huge*\n",
    "2.  Some operations are genuinely hard to do in parallel (e.g. sort)\n",
    "\n",
    "Additionally, some important operations like ``set_index`` work, but are slower\n",
    "than in Pandas because they include substantial inter-process communication and may write out to disk.\n",
    "\n",
    "Finally, `dask.dataframe` is quite new and non-trivial bugs are frequently reported (and quickly fixed)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### What definitely works?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Trivially parallelizable operations (fast):\n",
    "    *  Elementwise operations:  ``df.x + df.y``\n",
    "    *  Row-wise selections:  ``df[df.x > 0]``\n",
    "    *  Loc:  ``df.loc[4.0:10.5]``\n",
    "    *  Common aggregations:  ``df.x.max()``\n",
    "    *  Is in:  ``df[df.x.isin([1, 2, 3])]``\n",
    "    *  Datetime/string accessors:  ``df.timestamp.month``\n",
    "* Cleverly parallelizable operations (also fast):\n",
    "    *  groupby-aggregate (with common aggregations): ``df.groupby(df.x).y.max()``\n",
    "    *  value_counts:  ``df.x.value_counts``\n",
    "    *  Drop duplicates:  ``df.x.drop_duplicates()``\n",
    "    *  Join on index:  ``dd.merge(df1, df2, left_index=True, right_index=True)``\n",
    "* Operations requiring a shuffle (slow-ish, unless on index)\n",
    "    *  Set index:  ``df.set_index(df.x)``\n",
    "    *  groupby-apply (with anything):  ``df.groupby(df.x).apply(myfunc)``\n",
    "    *  Join not on the index:  ``pd.merge(df1, df2, on='name')``\n",
    "* Ingest operations\n",
    "    *  CSVs: ``dd.read_csv``\n",
    "    *  Pandas: ``dd.from_pandas``\n",
    "    *  Anything supporting numpy slicing: ``dd.from_array``\n",
    "    *  From any set of functions creating sub dataframes via ``dd.from_delayed``.\n",
    "    *  Dask.bag: ``mybag.to_dataframe(columns=[...])``"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "kernelspec": {
   "display_name": "Python [default]",
   "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}