01-Intro.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src='images/dask_horizontal.svg' width=500 align=left>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dask: Scaling Python Simply\n",
    "\n",
    "Dask is a distributed compute system for Python which scales efficiently from a single laptop up to thousands of servers. Dask is developed in ongoing collaboration with the PyData community so that it is easy to learn, integrate, and operate. Dask leverages regular Python code to scale the work you already do using the skills you already have.\n",
    "\n",
    "Dask is open source and freely available. It is developed in coordination with other community projects like Numpy, Pandas, and Scikit-Learn.\n",
    "\n",
    "* Built in Python\n",
    "* Scales *properly* from single laptops to 1000-node clusters\n",
    "* Leverages and interops with existing Python APIs as much as possible   \n",
    "\n",
    "### Why scale?\n",
    "\n",
    "In a nutshell,\n",
    "* In the 1900s, computer processors got faster: they ran more and more instructions per second\n",
    "* But, in the 2000s, due to a variety of engineering limitations, we can no longer acquire strictly faster processors. Instead we use more processors in collaboration\n",
    "* Many computations are faster if we can hold a dataset in local memory: while individual computers with huge memory do exist, it's usually easier and cheaper to use a collection of computers and \"pool\" their memory to store our data\n",
    "    \n",
    "#### Dask Ecosystem\n",
    "\n",
    "The Dask core library includes...\n",
    "* Dataframe\n",
    "* Array\n",
    "* Bag (multiset)\n",
    "* Delayed/Future\n",
    "* Scheduler\n",
    "* a few other cool tools\n",
    "\n",
    "Dask integrates with tons of libraries -- https://dask.org/#powered-by -- and can be used for tabular data, image data, geodata, simulation ... pretty much analytics or scientific computing task that makes sense in Python.\n",
    "\n",
    "In addition to the core Dask library and its Distributed scheduler, the Dask ecosystem connects several additional initiatives, including...\n",
    "* Dask ML - parallel machine learning, with a scikit-learn-style API\n",
    "* Dask-kubernetes and other deployment libraries\n",
    "* Dask-XGBoost\n",
    "* Dask-image\n",
    "* Dask-cuDF\n",
    "* ... and some others\n",
    "\n",
    "#### What's Not Part of Dask?\n",
    "\n",
    "There are lots of functions that integrate to Dask, but are not represented in the core Dask ecosystem, including...\n",
    "\n",
    "* ~~a SQL engine~~ -- check out https://github.com/nils-braun/dask-sql\n",
    "* data storage\n",
    "* data catalog\n",
    "* visualization\n",
    "* coarse-grained scheduling / orchestration\n",
    "* streaming\n",
    "\n",
    "... although there are typically other Python packages that fill these needs (e.g., Kartothek or Intake for a data catalog).\n",
    "\n",
    "### Our Agenda\n",
    "\n",
    "We'll focus on one particular, common use case for Dask: scaling Pandas dataframes to \n",
    "* larger datasets (which don't fit in memory) and \n",
    "* multiple processes (which could be on multiple nodes)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## What is it like to use Dask?\n",
    "\n",
    "There are 3 main ways that people use Dask, and you can use any combination or all of them.\n",
    "\n",
    "We'll take a quick test drive and see each of the 3 approaches\n",
    "* One-liners (or sometimes \"zero-liners\") where a tool already has Dask integration built in\n",
    "* Dask large-scale datastructures like Dask Dataframe: a scalable Pandas dataframe\n",
    "* Parallelizing custom computation: use the Dask engine to power your own code\n",
    "\n",
    "Let's spin up a small cluster to work with:"
   ]
  },
  {
   "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",
    "\n",
    "client = Client(cluster)\n",
    "\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Congratulations, you've started your first Dask cluster and you're using the state-of-the-art distributed scheduler, which is a great choice even though this cluster is on our local (front-end) machine right now.\n",
    "\n",
    "But how can we be sure it's alive and has these specs?\n",
    "\n",
    "__Workers Dashboard__\n",
    "\n",
    "Let's take a look at the Workers dashboard panel\n",
    "* Click the Dask logo in the JupyterLab side toolbar\n",
    "* Click the Workers button\n",
    "\n",
    "You should get a tab with a live, animated chart that looks something like this:\n",
    "\n",
    "<img src='images/workers.png' width=701>\n",
    "\n",
    "You can drag/position/snap that in JupyterLab, so that it's visible while you're coding.\n",
    "\n",
    "What are these \"workers\"? Just regular Python processes!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ok, let's try a \"one-liner\" ML application using Dask. We'll run an example from TPOT, an AutoML tool, to classify the \"digits\" dataset from Scikit-Learn (a set of very low-resolution handwritten digits)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import tpot\n",
    "from tpot import TPOTClassifier\n",
    "from sklearn.datasets import load_digits\n",
    "from sklearn.model_selection import train_test_split\n",
    "\n",
    "digits = load_digits()\n",
    "\n",
    "X_train, X_test, y_train, y_test = train_test_split(digits.data, digits.target, test_size=0.5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tp = TPOTClassifier(\n",
    "    generations=2,\n",
    "    population_size=10,\n",
    "    cv=2,\n",
    "    n_jobs=-1,\n",
    "    random_state=0,\n",
    "    verbosity=0,\n",
    "    config_dict=tpot.config.classifier_config_dict_light,\n",
    "    use_dask=True,\n",
    ")\n",
    "tp.fit(X_train, y_train)\n",
    "\n",
    "# quick look at test-set accuracy\n",
    "\n",
    "sum(tp.predict(X_test) == y_test)/len(y_test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice that the only Dask code we explicitly wrote there was a *kwarg* `use_dask=True` \n",
    "\n",
    "Next, let's take a very quick look at one of the Dask datastructures -- a parallel dataframe.\n",
    "\n",
    "We'll look at some records from the Seattle library system"
   ]
  },
  {
   "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.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "loans.groupby('CheckoutYear')['Checkouts'].sum().compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For the last stop on our quick preview of Dask, let's parallelize some custom code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "\n",
    "def roll_die(sides):\n",
    "    return random.randint(1,sides)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Local (regular) Python to roll 4d6"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "results = map(roll_die, [6] * 4)\n",
    "\n",
    "print(list(results))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using our Dask cluster to roll 4d6 in parallel:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dask\n",
    "\n",
    "roll_die = dask.delayed(roll_die)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster_work = map(roll_die, [6] * 4)\n",
    "\n",
    "dask.compute(*cluster_work)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Shut down our cluster for now"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.close()\n",
    "cluster.close()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Remote (or cloud) clusters vs. local cluster\n",
    "\n",
    "In that example, we ran a local cluster. But creating a Dask cluster on a cluster manager (like Kubernetes) or a public cloud (like AWS) isn't very different.\n",
    "\n",
    "One way to start a cluster on Kubernetes looks like\n",
    "\n",
    "```python\n",
    "cluster = KubeCluster.from_yaml('worker-spec.yml')\n",
    "client = Client(cluster)\n",
    "```\n",
    "\n",
    "and one way to start a cluster on AWS Fargate (container service) looks like\n",
    "\n",
    "```python\n",
    "cluster = FargateCluster(image=\"<hub-user>/<repo-name>[:<tag>]\")\n",
    "client = Client(cluster)\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Where are the docs?\n",
    "\n",
    "We'll provide more links to documentation as we go, but here's a quick list you can refer to:\n",
    "* Main project page https://dask.org/\n",
    "* Core documentation https://docs.dask.org/en/latest/\n",
    "* Distributed (scheduler) https://distributed.dask.org/en/latest/\n",
    "* Machine learning https://ml.dask.org/\n",
    "* Deployment tools\n",
    "    * Kubernetes https://kubernetes.dask.org/en/latest/\n",
    "    * AWS or Azure https://cloudprovider.dask.org/en/latest/\n",
    "    * YARN https://yarn.dask.org/en/latest/\n",
    "    \n",
    "## Some more Dask Community resources\n",
    "\n",
    "* Dask issues and source code https://github.com/dask\n",
    "* StackOverflow https://stackoverflow.com/questions/tagged/dask\n",
    "* Gitter https://gitter.im/dask/dask\n",
    "\n",
    "## How Do We Set Up and/or Deploy Dask?\n",
    "\n",
    "The easiest way to install Dask is with Anaconda: `conda install dask`\n",
    "\n",
    "__Schedulers and Clustering__\n",
    "\n",
    "Dask has a simple default scheduler called the \"single machine scheduler\" -- this is the scheduler that's used if your `import dask` and start running code without explicitly using a `Client` object. It can be handy for quick-and-dirty testing, but I would (*warning! opinion!*) suggest that a best practice is to __use the distributed scheduler even for single-machine workloads__\n",
    "\n",
    "The distributed scheduler can work with \n",
    "* threads (although that is often not a great idea due to the GIL) in one process\n",
    "* multiple processes on one machine\n",
    "* multiple processes on multiple machines\n",
    "\n",
    "The distributed scheduler has additional useful features including data locality awareness and realtime graphical dashboards.\n",
    "\n",
    "__Large-Scale and Cloud Deployment__\n",
    "\n",
    "Check out links under https://docs.dask.org/en/latest/setup.html for info on deploying to...\n",
    "* Clouds\n",
    "* HPC environments\n",
    "* Hadoop/YARN\n",
    "* Kubernetes\n",
    "\n",
    "... and more"
   ]
  },
  {
   "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
}