02-Analytics.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Analytics with Dask Dataframe"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client, LocalCluster\n",
    "\n",
    "cluster = LocalCluster(n_workers=2, threads_per_worker=1, memory_limit='700MiB')\n",
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we'll use Dask dataframe to access the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask.dataframe as ddf\n",
    "\n",
    "loans = ddf.read_parquet('data/checkouts-micro')\n",
    "loans"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__This looks a bit different from a Pandas dataframe ... so:__\n",
    "\n",
    "## What *is* a Dask dataframe?\n",
    "\n",
    "A Dask dataframe is a (virtual) collection of Pandas dataframes, divided along the index. You can picture it like this:\n",
    "\n",
    "<img src='images/dask-dataframe.svg' width=400>\n",
    "\n",
    "The smaller Pandas dataframes which make up the larger, virtual Dask dataframe, are called *partitions*\n",
    "\n",
    "So, at the top of the following output, the label __npartitions=__ refers to the number of constituent Pandas dataframes. You'll notice that Dask automatically chose a number of partitions to use, although you can customize that if you want to."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You won't often need to interact with individual partitions, but you can if you need to:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans.partitions[2]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Wait ... I thought you said the partition was a __Pandas__ dataframe!\n",
    "\n",
    "It is ... but we haven't computed it yet.\n",
    "\n",
    "In order to minimize extra computation, data movement, memory, and time, Dask's data structures try to be *lazy*\n",
    "\n",
    "This allows them to optimize their operation: for example, maybe you end up needing just 2 columns out of a 900-column-wide table ... in that case, it makes sense to see what's really needed before loading all of the data\n",
    "\n",
    "When we want to materialize a local, Python object, we add `.compute()` to API call\n",
    "\n",
    "So, to tell Dask that we want to load up that partition locally, we could type"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans.partitions[2].compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That looks a lot like Pandas output, but we can check to be sure:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(loans.partitions[2].compute())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can use some \"regular\" Python operators as well"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(loans)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Wait, wait ... I thought you just said I needed `.compute()`!\n",
    "\n",
    "#### __When do I need `.compute()` and how can I tell?__\n",
    "\n",
    "__Do call__ `.compute()` when you want a full Pandas object -- Dataframe or Series -- calculated for you *and* you want it loaded up in your local Python process (where your `Client` object lives).\n",
    "\n",
    "This is typical for small, report-type outputs, like we'll do later in this notebook.\n",
    "\n",
    "__Don't__ call `.compute()` on a huge Dask dataframe, because it likely won't fit in local memory anyway\n",
    "\n",
    "__Don't__ call `.compute()` if the goal is to write out a large dataset (perhaps one that you've transformed) to disk. There are APIs for doing that directly from the cluster, in parallel, so that your local process doesn't have to deal with all that data.\n",
    "\n",
    "__Don't__ call `.compute()` if there are simpler APIs designed for human, interactive consumption that might be more efficient, like `.head()` or `len()`\n",
    "\n",
    "Having a local result is convenient, but if we are generating large results, we may want (or need) to produce output in parallel to the filesystem, instead. \n",
    "\n",
    "There are writing counterparts to read methods which we can use:\n",
    "\n",
    "- `read_csv` \\ `to_csv`\n",
    "- `read_hdf` \\ `to_hdf`\n",
    "- `read_json` \\ `to_json`\n",
    "- `read_parquet` \\ `to_parquet`\n",
    "- and more ... you can bookmark https://docs.dask.org/en/latest/dataframe-api.html for the full API"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## What can I do with Dask dataframe and how do I do it?\n",
    "\n",
    "If you're used to Pandas, it takes a little adjustment to get used to working with data without seeing all those nice rows and columns on the screen. But most of the operations you're used to -- selecting and transforming columns, filtering rows, grouping and aggregating -- still work.\n",
    "\n",
    "In our first library task, we need to throw out the __Publisher__ column as well as \"old\" data.\n",
    "\n",
    "First, let's drop the columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans2 = loans.drop(columns=['Publisher'])\n",
    "loans2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "But what is \"old\" data? Let's find all of the years in the dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans2['CheckoutYear'].unique()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Hmm... this is a \"lazy\" Dask Series, but we really want the actual, concrete Series of unique years.\n",
    "We know this will be a small collection, so __it's time for `.compute()`__"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans.CheckoutYear.unique().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's keep records from 2010 onward, and we omit the 2020 data since it's ... anomalous. \n",
    "\n",
    "So we can filter that whole dataset using a Pandas-style filter"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans3 = loans2[(loans2['CheckoutYear'] >= 2010) & (loans2['CheckoutYear'] < 2020)]\n",
    "len(loans3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "But it's more *memory* efficient to use Pandas' `query`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans3 = loans3.query('CheckoutYear >= 2010 & CheckoutYear < 2020')\n",
    "len(loans3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, imagine we want to *drop all incomplete records* and then write out the cleaned, post-2010 dataset in Apache Parquet format.\n",
    "\n",
    ">\n",
    "> __Aside: Why Apache Parquet?__\n",
    ">\n",
    "> Apache Parquet is one of the most popular, efficient, and performant formats for large-scale structured data. \n",
    ">\n",
    "> Why? Because Parquet is a compressed, self-describing, binary *columnar* data format, which means that each column is stored apart from the others. So when we need to query just a few columns in a wide table, we can physically access just the ones we need on disk. The fastest data to process is the data you never load in the first place!\n",
    ">\n",
    "> Moreover, if we know what sorts of queries we will need to perform, we can *partition* by those values on disk as well. In our case, since we're partitioning by CheckoutYear, if we subsequently need records from 2016, we can access those and only those directly from the disk.\n",
    ">\n",
    "\n",
    "Let's `dropna()` and write out our data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans3.dropna().to_parquet('cleaned-loans', write_index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__Progress Dashboard__\n",
    "\n",
    "That seems to take a little while (well, a few seconds, at least). Let's open another dashboard view that will let us track progress.\n",
    "\n",
    "From the Dask dashboard palette, click `Progress` and drag that to snap at the bottom of the JupyterLab window.\n",
    "\n",
    "<img src='images/progress.png' width=901>\n",
    "\n",
    "You should see several colored progress bars. The colors correspond to specific functions being run (when those are functions you've defined, they'll show your function names; in this case, they're function names from the Dask dataframe library)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's look at trends in physical vs. digital loans (\"UsageClass\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "report = ddf.read_parquet('cleaned-loans/').groupby(['CheckoutYear', 'UsageClass']).agg({'Checkouts': 'sum'}).compute()\n",
    "report"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That may have stalled ... when Dask just stalls, it is usually a memory starvation issue, and the first things to try are \n",
    "* smaller partitions\n",
    "* more memory for your workers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "report = ddf.read_parquet('cleaned-loans/').repartition(50).groupby(['CheckoutYear', 'UsageClass']).agg({'Checkouts': 'sum'}).compute()\n",
    "report"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And since this is just a Pandas dataframe, we can plot it"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "report.unstack(level=1).plot()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "__Task Stream Dashboard__\n",
    "\n",
    "Let's look at one more dashboard. From the palette, choose Task Stream, and snap it somewhere convenient.\n",
    "\n",
    "With the Task Stream open, re-run the previous reports. Zoom in to where you can see individual tasks across your cluster cores -- color coded to match the other views like Progress -- as well as time spent transferring data (the red \"overlay\" boxes\")\n",
    "\n",
    "<img src='images/taskstream.png' width=901>\n",
    "\n",
    "This sort of X-ray vision into what's happening in the cluster makes tuning and troubleshooting a lot easier than doing so with log messages and summary stats."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Additional key features of Dask and Dask dataframe\n",
    "\n",
    "### Caching\n",
    "\n",
    "One benefit of using a cluster is having more processing power (cores). But equally valuable is having an expanded pool of memory: for example, most of us don't have 250GB of RAM in our laptop, while even a small cluster is likely to have that much memory available.\n",
    "\n",
    "To materialize a Dask dataframe (or any Delayed object) in the distributed RAM of the cluster, we use the `.persist()` API\n",
    "\n",
    "`.persist` is not lazy: it immediately starts working ... but it returns a Delayed right away because we work is not done yet. So we still get a token or handle. And, actually, a token is what we want: the whole point is that we want the big data in the cluster, not in our local Python runtime!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans_cached = ddf.read_parquet('cleaned-loans').persist()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can run some queries or transformations over the data in memory... or can we?\n",
    "How do we know if the data is loaded up yet?\n",
    "\n",
    "There are several ways!\n",
    "\n",
    "First, we can look at the __Graph Dashboard__: from the dashboard palette, click \"Graph\"\n",
    "\n",
    "Each Task (delayed Python function) gets a little square, and the key explains the color coding: red boxes are tasks whose result is stored in memory. \n",
    "\n",
    "For a big job (and a huge graph), we can watch the boxes turn red in real time ... a sort of RAM-storage progress bar.\n",
    "\n",
    "We can also access the information programmatically."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask\n",
    "\n",
    "dask.distributed.futures_of(loans_cached)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A Future is another kind of handle (similar to Promises in some languages) representing a task that was started but may not have finished (or may have failed altogether). In this example, we can see that each of the Futures is `finished`. \n",
    "\n",
    "We can also wait for all of them:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "dask.distributed.wait(loans_cached)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Our queries should be faster with the data in memory"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "loans_cached.mean().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Compare to the non-cached timing:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "ddf.read_parquet('cleaned-loans').mean().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In practice, your speedups will depend on how expensive the I/O is relative to the computation. \n",
    "\n",
    "* The slower, larger, and more distant and more expensive it is to process the source data, the more of an improvement you'll see.\n",
    "\n",
    "* On the other hand, the more expensive and complex your computation is, the less improvement you'll see.\n",
    "\n",
    "### Custom Functions with `.apply`\n",
    "\n",
    "Often, you'll want to apply your own logic to data in a Dask dataframe. Like Pandas, Dask supports the `.apply` method to run your own code over rows of data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def my_custom_length(field):\n",
    "    return len(field)\n",
    "\n",
    "loans_cached.Title.apply(my_custom_length)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It runs, but does suggest we add some schema information to help out."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans_cached.Title.apply(my_custom_length, meta=('special_feature', 'int64')).head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also apply to rows, allowing us to perform calculations or transformations depending on multiple columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def my_combo_length(fields):\n",
    "    return len(fields[0]) + len(fields[1])\n",
    "\n",
    "loans_cached[['Title', 'Subjects']].dropna().apply(my_combo_length, axis=1, meta=(None, 'int64')).head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Like Pandas, we can assign arbitrary columns to our dataframe:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "my_special_feature = loans_cached.Title.apply(my_custom_length, meta=('special_feature', 'int64'))\n",
    "loans_cached['my_special_feature'] = my_special_feature\n",
    "\n",
    "loans_cached.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Custom functions are also supported for aggregations and rolling (\"window\") computations.\n",
    "\n",
    "### Quirks and limitations\n",
    "\n",
    "As you've probably noticed, Dask dataframe implements a lot of the Pandas API. At the same time, there are also some quirks to get used to (e.g., the schema hints we just provided) as well as functionality that is not implemented ... at least not yet.\n",
    "\n",
    "You can refer to the docs to see which APIs are implemented differently (or not at all). But another approach is to try your planned computation (based on Pandas knowledge) on a small subset of your data -- in a non-destructive way -- and see if it runs and the results check out. Different users will likely prefer one or the other of these tactices.\n",
    "\n",
    "### Best practices\n",
    "\n",
    "Some additional best practices for working with Dask dataframe as well as patterns/anti-patterns are documented here\n",
    "* https://docs.dask.org/en/latest/dataframe-best-practices.html\n",
    "* https://docs.dask.org/en/latest/dataframe.html#common-uses-and-anti-uses\n",
    "\n",
    "Common scenarios are explained in the docs, including...\n",
    "* Shuffles https://docs.dask.org/en/latest/dataframe-groupby.html\n",
    "* Joins https://docs.dask.org/en/latest/dataframe-joins.html\n",
    "* Categorical types https://docs.dask.org/en/latest/dataframe-categoricals.html\n",
    "\n",
    "And if you're curious about how it all works, a design description of the internals is at https://docs.dask.org/en/latest/dataframe-design.html ... from there you can take a look at source if you'd still like see more.\n",
    "\n",
    "### Swappable partition dataframe implementations and RAPIDS cuDF\n",
    "\n",
    "Since Dask dataframe is architected around proxying to Pandas dataframes ...\n",
    "and Python allows us to swap in alternative objects, provided they implement the same protocol or interface (\"duck typing\") ...\n",
    "we can use Dask with other dataframe implementations.\n",
    "\n",
    "Most notably, this support scalable GPU-based dataframes but placing Dask on top of cuDF dataframes in NVIDIA RAPIDS\n",
    "* https://docs.rapids.ai/api/cudf/stable/10min.html#\n",
    "* https://docs.rapids.ai/api/cudf/stable/dask-cudf.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.close()\n",
    "cluster.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.8.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}