getting_started_notebooks/intro_tutorials/04_Introduction_to_Dask_using_cuDF_DataFrames.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"introduction\"></a>\n",
    "## Introduction to Dask using cuDF DataFrames\n",
    "#### By Paul Hendricks\n",
    "-------\n",
    "\n",
    "In this notebook, we will show how to work with cuDF DataFrames using Dask.\n",
    "\n",
    "**Table of Contents**\n",
    "\n",
    "* [Introduction to Dask using cuDF DataFrames](#introduction)\n",
    "* [Setup](#setup)\n",
    "* [Using cuDF DataFrames with Dask](#using)\n",
    "* [Conclusion](#conclusion)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"setup\"></a>\n",
    "## Setup\n",
    "\n",
    "This notebook was tested using the following Docker containers:\n",
    "\n",
    "* `rapidsai/rapidsai-dev-nightly:0.10-cuda10.0-devel-ubuntu18.04-py3.7` container from [DockerHub](https://hub.docker.com/r/rapidsai/rapidsai-nightly)\n",
    "\n",
    "This notebook was run on the NVIDIA GV100 GPU. Please be aware that your system may be different and you may need to modify the code or install packages to run the below examples. \n",
    "\n",
    "If you think you have found a bug or an error, please file an issue here: https://github.com/rapidsai/notebooks-contrib/issues\n",
    "\n",
    "Before we begin, let's check out our hardware setup by running the `nvidia-smi` command."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!nvidia-smi"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, let's see what CUDA version we have:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!nvcc --version"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Install graphviz\n",
    "The visualizations in this notebook require graphviz.  Your environment may not have it installed, but don't worry! If you don't, we're going to install it now.  This can take a little while, so sit tight."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "try:\n",
    "    import graphviz\n",
    "except ModuleNotFoundError:\n",
    "    os.system('apt update')\n",
    "    os.system('apt install -y graphviz')\n",
    "    os.system('conda install -c conda-forge graphviz -y')\n",
    "    os.system('conda install -c conda-forge python-graphviz -y')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's start by creating a local cluster of workers and a client to interact with that cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client\n",
    "from dask_cuda import LocalCUDACluster\n",
    "\n",
    "\n",
    "# create a local CUDA cluster\n",
    "cluster = LocalCUDACluster()\n",
    "client = Client(cluster)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll define a function called `load_data` that will create a `cudf.DataFrame` with two columns, `key` and `value`. The column `key` will be randomly filled with either a 0 or a 1, with 50% probability of either number being selected. The column `value` will be randomly filled with numbers sampled from a normal distribution."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import cudf; print('cuDF Version:', cudf.__version__)\n",
    "import numpy as np; print('NumPy Version:', np.__version__)\n",
    "\n",
    "\n",
    "def load_data(n_rows):\n",
    "    df = cudf.DataFrame()\n",
    "    random_state = np.random.RandomState(43210)\n",
    "    df['key'] = random_state.binomial(n=1, p=0.5, size=(n_rows,))\n",
    "    df['value'] = random_state.normal(size=(n_rows,))\n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll also define a function `head` that takes a `cudf.DataFrame` and returns the first 5 rows."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def head(dataframe):\n",
    "    return dataframe.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll define the number of workers as well as the number of rows each dataframe will have."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# define the number of workers\n",
    "n_workers = 4  # feel free to change this depending on how many GPUs you have\n",
    "\n",
    "# define the number of rows each dataframe will have\n",
    "n_rows = 125000000  # we'll use 125 million rows in each dataframe"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll create each dataframe using the `delayed` operator. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.delayed import delayed\n",
    "\n",
    "\n",
    "# create each dataframe using a delayed operation\n",
    "dfs = [delayed(load_data)(n_rows) for i in range(n_workers)]\n",
    "dfs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We see the result of this operation is a list of `Delayed` objects. It's important to note that these operations are \"delayed\" - nothing has been computed yet, meaning our data has not yet been created!\n",
    "\n",
    "We can apply the `head` function to each of our \"delayed\" dataframes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "head_dfs = [delayed(head)(df) for df in dfs]\n",
    "head_dfs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As before, we see that the result is a list of `Delayed` objects - an important thing to note is that our \"key\", or unique identifier for each operation, has changed. You should see the name of the function `head` followed by a hash sign. For example, one might see:\n",
    "\n",
    "```\n",
    "[Delayed('head-8e946db2-feaf-4e79-99ab-f732b6e28461'),\n",
    " Delayed('head-eb06bc77-9d5c-4a47-8c01-b5b36710b727'),\n",
    " Delayed('head-e1c976c8-3f94-4a01-8300-41def5117f93'),\n",
    " Delayed('head-7d0a7201-a973-4846-a68f-cb6f85b25076')]\n",
    "```\n",
    "\n",
    "Again, nothing has been computed - let's compute the results and execute the workflow using the `client.compute()` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import wait\n",
    "\n",
    "\n",
    "# use the client to compute - this means create each dataframe and take the head\n",
    "futures = client.compute(head_dfs)\n",
    "wait(futures)  # this will give Dask time to execute the work before moving to any subsequently defined operations\n",
    "futures"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We see that our results are a list of futures. Each object in this list tells us a bit information about itself: the status (pending, error, finished), the type of the object, and the key (unique identifief).\n",
    "\n",
    "We can use the `client.gather` method to collect the results of each of these futures."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# collect the results\n",
    "results = client.gather(futures)\n",
    "results"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We see that our results are a list of cuDF DataFrames, each having 2 columns and 5 rows. Let's inspect the first dataframe:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# let's inspect the head of the first dataframe\n",
    "print(results[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Voila! \n",
    "\n",
    "That was a pretty simple example. Let's see how we can use this perform a more complex operation like figuring how many total rows we have across all of our dataframes. We'll define a function called `length` that will take a `cudf.DataFrame` and return the first value of the `shape` attribute i.e. the number of rows for that particular dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def length(dataframe):\n",
    "    return dataframe.shape[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll define our operation on the dataframes we've created:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "lengths = [delayed(length)(df) for df in dfs]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And then use Python's built-in `sum` function to sum all of these lengths."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total_number_of_rows = delayed(sum)(lengths)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At this point, `total_number_of_rows` hasn't been computed yet. But we can still visualize the graph of operations we've defined using the `visualize()` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total_number_of_rows.visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The graph can be read from bottom to top. We see that for each worker, we will first execute the `load_data` function to create each dataframe. Then the function `length` will be applied to each dataframe; the results from these operations on each worker will then be combined into a single result via the `sum` function. \n",
    "\n",
    "Let's now execute our workflow and compute a value for the `total_number_of_rows` variable."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# use the client to compute the result and wait for it to finish\n",
    "future = client.compute(total_number_of_rows)\n",
    "wait(future)\n",
    "future"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We see that our computation has finished - our result is of type `int`. We can collect our result using the `client.gather()` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# collect result\n",
    "result = client.gather(future)\n",
    "result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "That's all there is to it! We can define even more complex operations and workflows using cuDF DataFrames by using the `delayed`, `wait`, `client.submit()`, and `client.gather()` workflow.\n",
    "\n",
    "However, there can sometimes be a drawback from using this pattern. For example, consider a common operation such as a groupby - we might want to group on certain keys and aggregate the values to compute a mean, variance, or even more complex aggregations. Each dataframe is located on a different GPU - and we're not guaranteed that all of the keys necessary for that groupby operation are located on a single GPU i.e. keys may be scattered across multiple GPUs. \n",
    "\n",
    "To make our problem even more concrete, let's consider the simple operation of grouping on our `key` column and calculating the mean of the `value` column. To sovle this problem, we'd have to sort the data and transfer keys and their associated values from one GPU to another - a tricky thing to do using the delayed pattern. In the example below, we'll show an example of this issue with the delayed pattern and motivate why one might consider using the `dask_cudf` API.\n",
    "\n",
    "First, let's define a function `groupby` that takes a `cudf.DataFrame`, groups by the `key` column, and calculates the mean of the `value` column."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def groupby(dataframe):\n",
    "    return dataframe.groupby('key')['value'].mean()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll apply the function `groupby` to each dataframe using the `delayed` operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "groupbys = [delayed(groupby)(df) for df in dfs]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We'll then execute that operation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# use the client to compute the result and wait for it to finish\n",
    "groupby_dfs = client.compute(groupbys)\n",
    "wait(groupby_dfs)\n",
    "groupby_dfs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = client.gather(groupby_dfs)\n",
    "results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for i, result in enumerate(results):\n",
    "    print('cuDF DataFrame:', i)\n",
    "    print(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This isn't exactly what we wanted though - ideally, we'd get one dataframe where for each unique key (0 and 1), we get the mean of the `value` column.\n",
    "\n",
    "We can use the `dask_cudf` API to help up solve this problem. First we'll import the `dask_cudf` library and then use the `dask_cudf.from_delayed` function to convert our list of delayed dataframes to an object of type `dask_cudf.core.DataFrame`. We'll use this object - `distributed_df` - along with the `dask_cudf` API to perform that \"tricky\" groupby operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask_cudf; print('Dask cuDF Version:', dask_cudf.__version__)\n",
    "\n",
    "\n",
    "# create a distributed cuDF DataFrame using Dask\n",
    "distributed_df = dask_cudf.from_delayed(dfs)\n",
    "print('Type:', type(distributed_df))\n",
    "distributed_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `dask_cudf` API closely mirrors the `cuDF` API. We can use a groupby similar to how we would with cuDF - but this time, our operation is distributed across multiple GPUs!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "result = distributed_df.groupby('key')['value'].mean().compute()\n",
    "result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lastly, let's examine our result!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"conclusion\"></a>\n",
    "## Conclusion\n",
    "\n",
    "In this notebook, we showed how to work with cuDF DataFrames using Dask.\n",
    "\n",
    "To learn more about RAPIDS, be sure to check out: \n",
    "\n",
    "* [Open Source Website](http://rapids.ai)\n",
    "* [GitHub](https://github.com/rapidsai/)\n",
    "* [Press Release](https://nvidianews.nvidia.com/news/nvidia-introduces-rapids-open-source-gpu-acceleration-platform-for-large-scale-data-analytics-and-machine-learning)\n",
    "* [NVIDIA Blog](https://blogs.nvidia.com/blog/2018/10/10/rapids-data-science-open-source-community/)\n",
    "* [Developer Blog](https://devblogs.nvidia.com/gpu-accelerated-analytics-rapids/)\n",
    "* [NVIDIA Data Science Webpage](https://www.nvidia.com/en-us/deep-learning-ai/solutions/data-science/)"
   ]
  },
  {
   "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}