Merge branch 'dask-docs' into 'master'

Add Dask documentation

See merge request !296

.spelling

@@ -804,3 +804,5 @@ npm
  - node_modules/spawn-sync/README.md
 iojs
 >>>>>>> readme
+UCX
+Dask-ssh

docs.it4i/software/data-science/dask.md

+# Dask
+
+[Dask](https://docs.dask.org/en/latest/) is a popular open-source library that allows you to
+parallelize your Python code using a distributed cluster easily. It can parallelize arbitrary
+Python code in the form of a task DAG (Directed Acyclic Graph), but it also offers parallelized
+versions of popular Python data science libraries like
+[numpy](https://docs.dask.org/en/latest/array.html) or
+[Pandas](https://docs.dask.org/en/latest/dataframe.html).
+
+## Installation
+
+To install Dask, load a recent version of Python 3 and install Dask using `pip`. We heavily
+recommend you to install Python packages inside a Python virtual environment.
+
+```bash
+# Load Python (preferably use a specific Python version)
+$ ml Python3
+
+# Create a virtual environment
+$ python3 -m venv dask
+
+# Activate and update the virtual environment
+$ source dask/bin/activate
+(dask) $ pip install -U setuptools pip wheel
+
+# Install Dask
+(dask) $ pip install distributed
+```
+
+## Starting a Dask Cluster
+
+Before you can use Dask, you need to set up a Dask cluster, which consists of a single server
+component which coordinates the cluster, and an arbitrary number of workers, which actually
+execute your tasks.
+
+![Dask cluster architecture](imgs/dask-arch.svg)
+
+After you start a PBS job, you should therefore first start the server and the workers on the
+available computing nodes and then run your Python program that uses Dask. There are multiple ways
+of deploying the cluster. A common scenario is to run a Dask server on a single computing node,
+run a single worker per node on all remaining computing nodes and then run your program on the node
+with the server.
+
+> There are some performance considerations to be taken into account regarding Dask cluster
+> deployment, see [below](#dask-performance-considerations) for more information.
+
+!!! note
+    All the following deployment methods assume that you are inside a Python environment that has
+    Dask installed. Do not forget to load Python and activate the correct virtual environment at
+    the beginning of your PBS job! And also do the same after connecting to any worker nodes
+    manually using SSH.
+
+### Manual Deployment
+
+Both the server and the worker nodes can be started using a CLI command. If you prefer manual
+deployment, you can manually start the server on a selected node and then start the workers on
+other nodes available inside your PBS job.
+
+```bash
+# Start the server on some node N
+$ dask-scheduler
+
+# Start a single worker on some other node, pass it the address of the server
+$ dask-worker tcp://<hostname-of-N>:8786
+```
+
+### Dask-ssh Deployment
+
+Dask actually contains [built-in support](https://docs.dask.org/en/latest/setup/ssh.html) for
+automating Dask deployment using SSH. It also supports nodefiles provided by PBS, so inside of your
+PBS job, you can simply run
+
+```bash
+$ dask-ssh --hostfile $PBS_NODEFILE
+```
+
+to start the Dask cluster on all available computing nodes. This will start the server on the first
+node of your PBS job and then a single worker on each node. The first node will therefore be shared
+by a server and a worker, which might not be ideal from a performance point of view.
+
+> Note that for this to work, the `paramiko` Python library has to be installed inside your Python
+> environment (you can install it using `$ pip install paramiko`).
+
+You can also start the Cluster directly from your
+[Python script](https://docs.dask.org/en/latest/setup/ssh.html#python-interface). In this way you
+can start the scheduler and the workers on separate nodes to avoid overcrowding the server node.
+
+### Other Deployment Options
+
+Dask has a lot of other ways of being deployed, e.g. using MPI, or using a shared file on the
+network file system. It also allows you to create a PBS job directly, wait for it to be started and
+then it starts the whole cluster inside the PBS job. You can find more information about Dask HPC
+deployment [here](https://docs.dask.org/en/latest/setup/hpc.html).
+
+## Connecting to the Cluster
+
+Once you have deployed your cluster, you must create a `Client` at the beginning of your program
+and pass it the address of the server.
+
+```python
+from distributed import Client
+
+client = Client("<hostname-of-server>:8786")
+```
+
+Once the client connects to the server successfully, all subsequent Dask computations will be
+parallelized using the cluster.
+
+Below are some examples of computations that you can perform with Dask. Note that the code should
+only be executed after a client was connected to a server!
+
+### Parallelize Arbitrary Python Code Using `Delayed`
+
+The `delayed` function (or a decorator) turns a Python function into a lazy computation. If you
+call such a function, it will not execute right away. It will only return a future object that can
+be composed with other futures to build a DAG of tasks. After you describe your whole computation,
+you can actually execute it using `dask.compute(<future>)`.
+
+```python
+import dask
+
+@dask.delayed
+def inc(x):
+    return x + 1
+
+@dask.delayed
+def double(x):
+    return x * 2
+
+@dask.delayed
+def add(x, y):
+    return x + y
+
+data = [1, 2, 3, 4, 5]
+
+output = []
+for x in data:
+    a = inc(x)
+    b = double(x)
+    c = add(a, b)
+    output.append(c)
+
+total = dask.delayed(sum)(output)
+
+# `total` is just a lazy computation
+# To get the actual value, call dask.compute
+result = dask.compute(total)
+```
+
+You can find more information about the `delayed` API
+[here](https://docs.dask.org/en/latest/delayed.html).
+
+### Parallelize `Pandas`
+
+Dask contains a module called `dataframe` which mirrors the API of `pandas`, a popular library for
+tabular analysis. Using `dataframe` you can distribute your `pandas` code to multiple nodes easily.
+You can find more information about it [here](https://docs.dask.org/en/latest/dataframe.html).
+
+Here is an example of its usage:
+
+```python
+import dask.dataframe as pd
+
+# Load CSV
+df = pd.read_csv("table.csv")
+
+# Describe a lazy computation (this is not computed yet)
+df2 = df[df.y == "a"].x + 1
+
+# Actually compute the table operations on worker nodes
+result = df2.compute()
+```
+
+### Parallelize `Numpy`
+
+Dask contains a module called `array` which mirrors the API of `numpy`, a popular library for
+n-dimensional array computations. Using `array` you can distribute your `numpy` code to multiple
+nodes easily. You can find more information about it
+[here](https://examples.dask.org/array.html).
+
+Here is an example of its usage:
+
+```python
+import dask.array as np
+x = np.random.random((10000, 10000), chunks=(1000, 1000))
+
+# Describe a lazy computation (this is not computed yet)
+y = x + x.T
+z = y[::2, 5000:].mean(axis=1)
+
+# Actually compute the arary operations on worker nodes
+result = z.compute()
+```
+
+## Dask Performance Considerations
+
+Dask should be fast enough by default for most use cases, but there are some considerations that
+should be taken into account.
+
+### Selecting the Number of Processes and Threads Per Worker
+
+When starting a Dask worker on a node, it will by default use a number of threads equal to the
+number of cores on the given machine. At the same time, (C)Python uses a global lock
+([GIL](https://realpython.com/python-gil/)) that prevents more than a single thread to execute at
+once. This is fine if your computational tasks are I/O bound or if they spend most of their time
+inside C libraries that release the GIL.
+
+However, if your tasks executed with Dask are heavily compute-bound, and they hold the GIL (e.g.
+the heavy computation is performed directly inside Python), you might not be able to fully harness
+all cores of the worker node.
+
+To solve this, you can run multiple workers (each with a reduced number of threads) per node. This
+is a trade-off. With more workers on a node, you will be able to utilize more cores (assuming
+your tasks are compute-bound). However, you will also increase the pressure on the central server
+and on the network, because there will be more workers that will communicate with each other and
+with the server.
+
+You can choose the number of workers and threads per each worker using the `--nprocs` and
+`--nthreads` parameters of `dask-worker` (there are similar arguments when using other deployment
+methods).
+
+Some examples (assuming a node with 24 cores):
+
+```bash
+# Run a single worker using 24 threads. Reduces network and server traffic, but may not utilize all cores.
+$ dask-worker --nprocs 1 --nthreads 24
+
+# Run 24 workers, each with a single thread. Maximizes core usage, but may overload server or network.
+$ dask-worker --nprocs 24 --nthreads 1
+
+# Run 6 workers, each with 4 threads. Strikes a balance between core usage and network/server pressure.
+$ dask-worker --nprocs 6 --nthreads 4
+```
+
+From our experiments, we found that often it is best to run a single worker per each core of a
+node to achieve the best performance. With this configuration, we found that Dask scales reasonably
+up to 200 workers (e.g. <10 Barbora nodes). If you start to run into performance problems because
+of the amount of workers, try to use [RSDS](#rsds) to achieve better scaling.
+
+### Memory Considerations
+
+A common reason to use Dask is that your computation does not fit inside the memory of a single
+node. Dask can alleviate this, but you still have to be careful about your memory usage. For
+example, when you use the `dataframe` or `array` API, you should pay attention into how many
+partitions (or chunks) is your data split it. If you use a single partition, it will probably take
+a lot of memory, and it will not offer many possibilities for parallelization. You can find more
+information [here](https://docs.dask.org/en/latest/dataframe-design.html#partitions).
+
+By default, multiple workers on a node will split the available memory. Therefore, if the node
+has 100 GiB of RAM and you start the Dask worker on a single node like this:
+
+```bash
+$ dask-worker --nprocs 10
+```
+
+Each worker will have around 10 GiB memory available. You can set the memory requirements of each
+worker manually using the `--memory-limit` argument.
+
+If your program is strictly memory-bound, you can also try alternative approaches to Dask.
+As an example, [Vaex](https://github.com/vaexio/vaex) is a library that allows you to easily
+process dataframes that do not fit inside your operating memory.
+
+### UCX
+
+In some cases (especially with many workers), the network can be a bottleneck. By default, Dask
+uses TCP/IP for communication, but it also has support for UCX, which enables more efficient usage
+of the available InfiniBand interfaces. It is a bit cumbersome to set up, but if you want to try,
+check [this tutorial](https://ucx-py.readthedocs.io/en/latest/dask.html).
+
+### Nvidia
+
+Dask has built-in support for GPU workers. You can find more information about this use case
+[here](https://docs.dask.org/en/latest/gpu.html).
+
+### RSDS
+
+If you need to run a large amount of tasks and Dask does not perform well enough, you can try to
+use [RSDS](https://github.com/It4innovations/rsds). It is our version of Dask which is optimized
+for HPC use cases, and it should provide better scaling than Dask. You can read more about RSDS
+in this [article](https://arxiv.org/abs/2010.11105).

mkdocs.yml

@@ -127,6 +127,8 @@ nav:
       - Orca: software/chemistry/orca.md
       - Phono3py: software/chemistry/phono3py.md
     - Compilers: software/compilers.md
+    - Data Science:
+        - Dask: software/data-science/dask.md
     - Debuggers:
       - Introduction: software/debuggers/introduction.md
       - Aislinn: software/debuggers/aislinn.md