03-Cluster_client.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 3. Starting and monitoring Dask"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A very nice feature of Dask, is that it allows to use the almost exact same code to perform distributed computing on a laptop, in the cloud or on a cluster. This is made possible through the ```dask.distributed``` module, which takes care of setting up two important components: the cluster and the client. The cluster represents the computing resources reserved (cores on a laptop, jobs on a cluster etc.) while the client is the interface that allows us to access to that cluster."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cluster creation\n",
    "First we create a cluster, here a local cluster on this machine, but it could also be a cluster e.g. on an HPC facility. We can specify many options here, in particular the number of workers (separate Python processes), the threads per worker etc. There are many options to configure computing resources that go beyond this introduction (see e.g. [here](https://distributed.dask.org/en/latest/worker.html) for more details)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import LocalCluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster = LocalCluster(n_workers=1, threads_per_worker=2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Connecting to cluster via client\n",
    "Now we can connect a client to our cluster to be able to acutally use it and submit computations. This is so to say our interface to the cluster:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import Client"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client = Client(cluster)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here is our client:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We see on the right a summary of the status of our cluster and on the left, two important addresses:\n",
    "\n",
    "- The dashboard address (of the type http://127.0.0.1:8787/status) leads us to a dashboard where we can monitor the activity of the cluster. The Dask extension for Jupyterlab allows us to have acces to the monitoring dashborad directly within Jupyterlab. We can just choose the Dask icon on the left and use the Dashboard address to access all panels. **Note that when using a Jupyterhub you have to change the address to https://address-of-hub.ch/user/your-user-name/proxy/8787**\n",
    " \n",
    "- The scheduler address allows us to create additional workers, e.g. directly in the terminal, by pointing them to the correct scheduler and typing:\n",
    "\n",
    "    ```\n",
    "    dask-worker \"tcp://127.0.0.1:55323\"\n",
    "    ```\n",
    "\n",
    "On top of many methods and attributes, the ```client``` also offers an interactive interface to the cluster:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client.cluster"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Here we can interactively select the number of workers and say if we want it to be automatically adjusted."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using Dask extension\n",
    "\n",
    "An alternative to create a cluster, is to use the Dask extension. If you click on \"+NEW\", a cluster is automatically created. To use it in your notebook simple click on the \"<>\" button which will add the client to your current notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "jupyter": {
     "source_hidden": true
    }
   },
   "outputs": [],
   "source": [
    "from IPython.display import HTML\n",
    "\n",
    "HTML(\"\"\"\n",
    "<video width=\"800\" controls>\n",
    "  <source src=\"images/cluster_extension.mp4\" type=\"video/mp4\">\n",
    "</video>\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example\n",
    "We are going to see all these pieces in action now using a simple example handling numpy arrays."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask import delayed\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "all_vals = []\n",
    "for i in range(10):\n",
    "    ar = delayed(np.random.randint)(1,100,(1000,1000,10))\n",
    "    ar = delayed(np.log)(ar)#[::2,::2,0]\n",
    "    mean_val = delayed(np.mean)(ar)\n",
    "    all_vals.append(mean_val)\n",
    "maxval = delayed(np.max)(all_vals)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "maxval.visualize()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "maxval.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we do the same operation for a lot more iterations:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "all_vals = []\n",
    "for i in range(1000):\n",
    "    ar = delayed(np.random.randint)(1,100,(1000,1000,10))\n",
    "    ar = delayed(np.log)(ar)#[::2,::2,0]\n",
    "    mean_val = delayed(np.mean)(ar)\n",
    "    all_vals.append(mean_val)\n",
    "maxval = delayed(np.max)(all_vals)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "maxval.compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since this is very slow, we can now stop the operation, add a worker and restart.\n",
    "\n",
    "Alternatively, as described above, to avoid stopping the calculation, we can dymamically assing new workers to the scheduler:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Asynchronous calculation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have just seen how to add new workers to a running computation. We had to do that through the command line, because the computation is blocking the execution of other tasks until completion. We can go around this by executing the tasks asynchronously using the ```client.compute()``` function instead of the regular ```compute()``` method. Asynchronous calculation is a complex topic about which you can learn more e.g. [here](https://distributed.dask.org/en/latest/manage-computation.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "value = client.compute(maxval)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "value.result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercise\n",
    "\n",
    "Open a new notebook and just try to create a cluster, a client, a dashboard, and run some code on it."
   ]
  },
  {
   "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.5"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": false,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {},
   "toc_section_display": true,
   "toc_window_display": true
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}