Dask logo

Dask Schedulers

In this tutorial, you learn:

  • Components of Dask Schedulers

  • Types of Dask Schedulers

  • Single Machine Schedulers

Related Documentation

Introduction

As we mentioned in our Dask overview, Dask is composed of two main parts:

  1. Dask Collections (APIs)

  2. Dynamic Task Scheduling

So far, we have talked about different Dask collections, but in this tutorial we are going to talk more about the second part.

The Dask scheduler - our task orchestrator

The Dask.distributed task scheduler is a centralized, dynamic system that coordinates the efforts of various dask worker processes spread accross different machines.

When a computational task is submitted, the Dask distributed scheduler sends it off to a Dask cluster - simply a collection of Dask workers. A worker is typically a separate Python process on either the local host or a remote machine.

To perform work, a scheduler must be assigned resources in the form of a Dask cluster. The cluster consists of the following components:

  • scheduler : A scheduler creates and manages task graphs and distributes tasks to workers.

  • workers : A worker is typically a separate Python process on either the local host or a remote machine. A Dask cluster usually consists of many workers. Basically, a worker is a Python interpretor which will perform work on a subset of our dataset.

  • client - A high-level interface that points to the scheduler (often local but not always). A client serves as the entry point for interacting with a Dask scheduler.

Dask Distributed Cluster

Image credit: Dask Contributors

Schedulers

Dask essentially offers two types of schedulers:

High level collections are used to generate task graphs which can be executed by schedulers on a single machine or a cluster.

Image credit: Dask Contributors

1. Single machine scheduler

  • The Single-machine Scheduler schedules tasks and manages the execution of those tasks on the same machine where the scheduler is running.

  • It is designed to be used in situations where the amount of data or the computational requirements are too large for a single process to handle, but not large enough to warrant the use of a cluster of machines.

  • It is relatively simple and cheap to use but it does not scale as it only runs on a single machine.

Single machine scheduler is the default choice used by Dask.

In Dask, there are several types of single machine schedulers that can be used to schedule computations on a single machine:

1.1. Single-threaded scheduler

This scheduler runs all tasks serially on a single thread.
This is only useful for debugging and profiling, but does not have any parallelization.

1.2. Threaded scheduler

The threaded scheduler uses a pool of local threads to execute tasks concurrently.
This is the default scheduler for Dask, and is suitable for most use cases on a single machine. Multithreading works well for Dask Array and Dask DataFrame.

To select one of the above scheduler for your computation, you can specify it when doing .compute():

For example:

this.compute(scheduler="single-threaded")  # for debugging and profiling only

As mentioned above the threaded scheduler is the default scheduler in Dask. But you can set the default scheduler to Single-threaded or multi-processing by:

import dask
dask.config.set(scheduler='synchronous')  # overwrite default with single-threaded scheduler

Multi-processing works well for pure Python code - delayed functions and operations on Dask Bags.

Let’s compare the performance of each of these single-machine schedulers:

import numpy as np
import dask.array as da

%%time
## - numpy performance
xn = np.random.normal(10, 0.1, size=(20_000, 20_000))
yn = xn.mean(axis=0)
yn

CPU times: user 8.43 s, sys: 236 ms, total: 8.66 s
Wall time: 8.66 s

array([ 9.99980945, 10.00060965, 10.00059974, ...,  9.99946232,
        9.99989715,  9.99832153])

%%time
# -- dask array using the default
xd = da.random.normal(10, 0.1, size=(20_000, 20_000), chunks=(2000, 2000))
yd = xd.mean(axis=0)
yd.compute()

CPU times: user 13.2 s, sys: 63.8 ms, total: 13.3 s
Wall time: 3.48 s

array([10.00046386, 10.00048034, 10.00063128, ...,  9.99887649,
       10.00112084,  9.99934577])

import time
# -- dask testing different schedulers:
for sch in ['threading', 'processes', 'sync']:
    t0 = time.time()
    r = yd.compute(scheduler=sch)
    t1 = time.time()
    print(f"{sch:>10} :  {t1 - t0:0.4f} s")

 threading :  3.3297 s

 processes :  4.1177 s

      sync :  8.5334 s

yd.dask

HighLevelGraph

HighLevelGraph with 4 layers and 240 keys from all layers.

Layer1: normal

normal-da61db80bcb20bfe79923f180a2d2379

layer_type MaterializedLayer
is_materialized True
number of outputs 100
shape (20000, 20000)
dtype float64
chunksize (2000, 2000)
type dask.array.core.Array
chunk_type numpy.ndarray
20000 20000

Layer2: mean_chunk

mean_chunk-461d4f1a565829a12a3bacc02e00fbd3

layer_type Blockwise
is_materialized True
number of outputs 100
shape (20000, 20000)
dtype float64
chunksize (2000, 2000)
type dask.array.core.Array
chunk_type numpy.ndarray
depends on normal-da61db80bcb20bfe79923f180a2d2379
20000 20000

Layer3: mean_combine-partial

mean_combine-partial-7550d65f4d76ee03c4b5a7331a31dc46

layer_type MaterializedLayer
is_materialized True
number of outputs 30
shape (3, 20000)
dtype float64
chunksize (1, 2000)
type dask.array.core.Array
chunk_type numpy.ndarray
depends on mean_chunk-461d4f1a565829a12a3bacc02e00fbd3
20000 3

Layer4: mean_agg-aggregate

mean_agg-aggregate-f6d36d2bf420e465a25b41909702a11e

layer_type MaterializedLayer
is_materialized True
number of outputs 10
shape (20000,)
dtype float64
chunksize (2000,)
type dask.array.core.Array
chunk_type numpy.ndarray
depends on mean_combine-partial-7550d65f4d76ee03c4b5a7331a31dc46
20000 1 
yd
Array Chunk
Bytes 156.25 kiB 15.62 kiB
Shape (20000,) (2000,)
Dask graph 10 chunks in 4 graph layers
Data type float64 numpy.ndarray
20000 1

  • Notice how sync scheduler takes almost the same time as pure NumPy code.

  • Why is the multiprocessing scheduler so much slower?

If you use the multiprocessing backend, all communication between processes still needs to pass through the main process because processes are isolated from other processes. This introduces a large overhead.

The Dask developers recommend using the Dask Distributed Scheduler which we will cover now.

2. Distributed scheduler

  • The Distributed scheduler or dask.distributed schedules tasks and manages the execution of those tasks on workers from a single or multiple machines.

  • This scheduler is more sophisticated and offers more features including a live diagnostic dashboard which provides live insight on performance and progress of the calculations.

In most cases, dask.distributed is preferred since it is very scalable, and provides and informative interactive dashboard and access to more complex Dask collections such as futures.

2.1. Local Cluster

A Dask Local Cluster refers to a group of worker processes that run on a single machine and are managed by a single Dask scheduler.

This is useful for situations where the computational requirements are not large enough to warrant the use of a full cluster of separate machines. It provides an easy way to run parallel computations on a single machine, without the need for complex cluster management or other infrastructure.

Let’s start by creating a Local Cluster

For this we need to set up a LocalCluster using dask.distributed and connect a client to it.

from dask.distributed import LocalCluster, Client

cluster = LocalCluster()
client = Client(cluster)
client

Client

Client-bd34ffa6-2371-11ef-8b14-00224802c87d

Connection method: Cluster object Cluster type: distributed.LocalCluster
Dashboard: http://127.0.0.1:8787/status

Cluster Info

☝️ Click the Dashboard link above.

👈 Or click the “Search” 🔍 button in the dask-labextension dashboard.

If no arguments are specified in LocalCluster it will automatically detect the number of CPU cores your system has and the amount of memory and create workers to appropriately fill that.

A LocalCluster will use the full resources of the current JupyterLab session. For example, if you used BinderHub, it will use the number of CPUs selected.

Note that LocalCluster() takes a lot of optional arguments, allowing you to configure the number of processes/threads, memory limits and other settings.

You can also find your cluster dashboard link using :

cluster.dashboard_link

'http://127.0.0.1:8787/status'

%%time
# -- dask array using the default
xd = da.random.normal(10, 0.1, size=(30_000, 30_000), chunks=(3000, 3000))
yd = xd.mean(axis=0)
yd.compute()

CPU times: user 523 ms, sys: 104 ms, total: 627 ms
Wall time: 8.72 s

array([10.00065933,  9.99971707, 10.00026382, ...,  9.99954561,
        9.99965514,  9.99968868])

Always remember to close your local Dask cluster:

client.shutdown()

Dask Distributed (Cluster Managers)

So far we have talked about running a job on a local machine.

Dask can be deployed on distributed infrastructure, such as a an HPC system or a cloud computing system.

High level collections are used to generate task graphs which can be executed by schedulers on a single machine or a cluster.

Image credit: Dask Contributors

Dask Cluster Managers have different names corresponding to different computing environments. Some examples are dask-jobqueue for your HPC systems (including PBSCluster) or Kubernetes Cluster for machines on the Cloud.

The NCAR tutorial series includes an in-depth exploration and practical use cases of Dask on HPC systems and best practices for Dask on HPC. For the complete set of NCAR tutorial materials, please refer to the main NCAR tutorial content available here.

For more information visit the Dask Docs.

On this page