getting_started_notebooks/basics/Getting_Started_with_Dask.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"introduction\"></a>\n",
    "## Introduction to Dask\n",
    "#### By Paul Hendricks\n",
    "-------\n",
    "\n",
    "In this notebook, we will show how to get started with Dask using basic Python primitives like integers and strings.\n",
    "\n",
    "**Table of Contents**\n",
    "\n",
    "* [Introduction to Dask](#introduction)\n",
    "* [Setup](#setup)\n",
    "* [Introduction to Dask](#dask)\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` 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": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!apt update\n",
    "!apt install -y graphviz\n",
    "!conda install graphviz\n",
    "!conda install python-graphviz"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"dask\"></a>\n",
    "## Introduction to Dask\n",
    "\n",
    "Dask is a library that allows for parallelized computing. Written in Python, it allows one to compose complex workflows using large data structures like those found in NumPy, Pandas, and cuDF. In the following examples and notebooks, we'll show how to use Dask with cuDF to accelerate common ETL tasks as well as build and train machine learning models like Linear Regression and XGBoost.\n",
    "\n",
    "To learn more about Dask, check out the documentation here: http://docs.dask.org/en/latest/\n",
    "\n",
    "#### Client/Workers\n",
    "\n",
    "Dask operates by creating a cluster composed of a \"client\" and multiple \"workers\". The client is responsible for scheduling work; the workers are responsible for actually executing that work. \n",
    "\n",
    "Typically, we set the number of workers to be equal to the number of computing resources we have available to us. For CPU based workflows, this might be the number of cores or threads on that particlular machine. For example, we might set `n_workers = 8` if we have 8 CPU cores or threads on our machine that can each operate in parallel. This allows us to take advantage of all of our computing resources and enjoy the most benefits from parallelization.\n",
    "\n",
    "On a system with one or more GPUs, we usually set the number of workers equal to the number of GPUs available to us. Dask is a first class citizen in the world of General Purpose GPU computing and the RAPIDS ecosystem makes it very easy to use Dask with cuDF and XGBoost. \n",
    "\n",
    "Before we get started with Dask, we need to setup a Local Cluster of workers to execute our work and a Client to coordinate and schedule work for that cluster. As we see below, we can inititate a `cluster` and `client` using only few lines of code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask; print('Dask Version:', dask.__version__)\n",
    "from dask.distributed import Client, LocalCluster\n",
    "import subprocess\n",
    "\n",
    "# parse the hostname IP address\n",
    "cmd = \"hostname --all-ip-addresses\"\n",
    "process = subprocess.Popen(cmd.split(), stdout=subprocess.PIPE)\n",
    "output, error = process.communicate()\n",
    "ip_address = str(output.decode()).split()[0]\n",
    "\n",
    "# create a local cluster with 4 workers\n",
    "n_workers = 4\n",
    "cluster = LocalCluster(ip=ip_address, n_workers=n_workers)\n",
    "client = Client(cluster)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's inspect the `client` object to view our current Dask status. We should see the IP Address for our Scheduler as well as the the number of workers in our Cluster. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# show current Dask status\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also see the status and more information at the Dashboard, found at `http://<ip_address>/status`. You can ignore this for now, we'll dive into this in subsequent tutorials.\n",
    "\n",
    "With our client and workers setup, it's time to execute our first program in parallel. We'll define a function that takes some value `x` and adds 5 to it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def add_5_to_x(x):\n",
    "    return x + 5"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next, we'll iterate through our `n_workers` and create an execution graph, where each worker is responsible for taking its ID and passing it to the function `add_5_to_x`. For example, the worker with ID 2 will take its ID and add 5, resulting in the value 7."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask import delayed\n",
    "\n",
    "addition_operations = [delayed(add_5_to_x)(i) for i in range(n_workers)]\n",
    "addition_operations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The above output shows a list of several `Delayed` objects. An important thing to note is that the workers aren't actually executing these results - we're just defining the execution graph for our client to execute later. The `delayed` function wraps our function `add_5_to_x` and returns a `Delayed` object. This ensures that this computation is in fact \"delayed\" - or lazily evaluated - and not executed on the spot i.e. when we define it.\n",
    "\n",
    "Next, let's sum each one of these intermediate results. We can accomplish this by wrapping Python's built-in `sum` function using our `delayed` function and storing this in a variable called `total`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total = delayed(sum)(addition_operations)\n",
    "total"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the `graphviz` library, we can use the `visualize` method of a `Delayed` object to visualize our current graph."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "total.visualize()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we mentioned before, none of these results - intermediate or final - have actually been compute. We can compute them using the `compute` method of our `client`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "addition_futures = client.compute(addition_operations, optimize_graph=False, fifo_timeout=\"0ms\")\n",
    "total_future = client.compute(total, optimize_graph=False, fifo_timeout=\"0ms\")\n",
    "time.sleep(1)  # this will give Dask time to execute each worker"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's inspect the output of each call to `client.compute`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "addition_futures"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can see from the above output that our `addition_futures` variable is a list of `Future` objects - not the \"actual results\" of adding 5 to each of `[0, 1, 2, 3]`. These `Future` objects are a promise that at one point a computation will take place and we will be left with a result. Dask is responsible for ensuring that promise by delegating that task to the appropriate Dask worker and collecting the result.\n",
    "\n",
    "Let's take a look at our `total_future` object:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(total_future)\n",
    "print(type(total_future))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Again, we see that this is an object of type `Future` as well as metadata about the status of the request (i.e. whether it has finished or not), the type of the result, and a key associated with that operation. To collect and print the result of each of these `Future` objects, we can call the `result()` method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "addition_results = [future.result() for future in addition_futures]\n",
    "print('Addition Results:', addition_results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we see the results that we want from our addition operations. We can also use the simpler syntax of the `client.gather` method to collect our results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "addition_results = client.gather(addition_futures)\n",
    "total_result = client.gather(total_future)\n",
    "print('Addition Results:', addition_results)\n",
    "print('Total Result:', total_result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Awesome! We just wrote our first distributed workflow.\n",
    "\n",
    "To confirm that Dask is truly executing in parallel, let's define a function that sleeps for 1 second and returns the string \"Success!\". In serial, this function should take our 4 workers around 4 seconds to execute."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def sleep_1():\n",
    "    time.sleep(1)\n",
    "    return 'Success!'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "for _ in range(n_workers):\n",
    "    sleep_1()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As expected, our process takes about 4 seconds to run. Now let's execute this same workflow in parallel using Dask."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%time\n",
    "\n",
    "# define delayed execution graph\n",
    "sleep_operations = [delayed(sleep_1)() for _ in range(n_workers)]\n",
    "\n",
    "# use client to perform computations using execution graph\n",
    "sleep_futures = client.compute(sleep_operations, optimize_graph=False, fifo_timeout=\"0ms\")\n",
    "\n",
    "# collect and print results\n",
    "sleep_results = client.gather(sleep_futures)\n",
    "print(sleep_results)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using Dask, we see that this whole process takes a little over a second - each worker is executing in parallel!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"conclusion\"></a>\n",
    "## Conclusion\n",
    "\n",
    "In this tutorial, we learned how to use Dask with basic Python primitives like integers and strings.\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
}