Command Line

This is the most fundamental way to deploy Dask on multiple machines. In production environments, this process is often automated by some other resource manager. Hence, it is rare that people need to follow these instructions explicitly. Instead, these instructions are useful for IT professionals who may want to set up automated services to deploy Dask within their institution.

A dask.distributed network consists of one dask-scheduler process and several dask-worker processes that connect to that scheduler. These are normal Python processes that can be executed from the command line. We launch the dask-scheduler executable in one process and the dask-worker executable in several processes, possibly on different machines.

To accomplish this, launch dask-scheduler on one node:

$ dask-scheduler
Scheduler at:   tcp://192.0.0.100:8786

Then, launch dask-worker on the rest of the nodes, providing the address to the node that hosts dask-scheduler:

$ dask-worker tcp://192.0.0.100:8786
Start worker at:  tcp://192.0.0.1:12345
Registered to:    tcp://192.0.0.100:8786

$ dask-worker tcp://192.0.0.100:8786
Start worker at:  tcp://192.0.0.2:40483
Registered to:    tcp://192.0.0.100:8786

$ dask-worker tcp://192.0.0.100:8786
Start worker at:  tcp://192.0.0.3:27372
Registered to:    tcp://192.0.0.100:8786

The workers connect to the scheduler, which then sets up a long-running network connection back to the worker. The workers will learn the location of other workers from the scheduler.

Handling Ports

The scheduler and workers both need to accept TCP connections on an open port. By default, the scheduler binds to port 8786 and the worker binds to a random open port. If you are behind a firewall then you may have to open particular ports or tell Dask to listen on particular ports with the --port and --worker-port keywords.:

dask-scheduler --port 8000
dask-worker --bokeh-port 8000 --nanny-port 8001

Nanny Processes

Dask workers are run within a nanny process that monitors the worker process and restarts it if necessary.

Diagnostic Web Servers

Additionally, Dask schedulers and workers host interactive diagnostic web servers using Bokeh. These are optional, but generally useful to users. The diagnostic server on the scheduler is particularly valuable, and is served on port 8787 by default (configurable with the --bokeh-port keyword).

Note

For more information about relevant ports, please take a look at the help pages with dask-scheduler --help and dask-worker --help

Automated Tools

There are various mechanisms to deploy these executables on a cluster, ranging from manually SSH-ing into all of the machines to more automated systems like SGE/SLURM/Torque or Yarn/Mesos. Additionally, cluster SSH tools exist to send the same commands to many machines. We recommend searching online for “cluster ssh” or “cssh”.

API

Warning

The command line documentation here may differ depending on your installed version. We recommend referring to the output of <command> --help.

dask-scheduler

dask-scheduler [OPTIONS] [PRELOAD_ARGV]...

Options

--host <host>

URI, IP or hostname of this server

--port <port>

Serving port

--interface <interface>

Preferred network interface like ‘eth0’ or ‘ib0’

--tls-ca-file <tls_ca_file>

CA cert(s) file for TLS (in PEM format)

--tls-cert <tls_cert>

certificate file for TLS (in PEM format)

--tls-key <tls_key>

private key file for TLS (in PEM format)

--bokeh-port <bokeh_port>

Deprecated. See –dashboard-address

--dashboard-address <dashboard_address>

Address on which to listen for diagnostics dashboard

--bokeh, --no-bokeh

Launch Bokeh Web UI [default: True]

--show, --no-show

Show web UI

--bokeh-whitelist <bokeh_whitelist>

IP addresses to whitelist for bokeh.

--bokeh-prefix <bokeh_prefix>

Prefix for the bokeh app

--use-xheaders <use_xheaders>

User xheaders in bokeh app for ssl termination in header [default: False]

--pid-file <pid_file>

File to write the process PID

--scheduler-file <scheduler_file>

File to write connection information. This may be a good way to share connection information if your cluster is on a shared network file system.

--local-directory <local_directory>

Directory to place scheduler files

--preload <preload>

Module that should be loaded by the scheduler process like “foo.bar” or “/path/to/foo.py”.

Arguments

PRELOAD_ARGV

Optional argument(s)

dask-worker

dask-worker [OPTIONS] [SCHEDULER] [PRELOAD_ARGV]...

Options

--tls-ca-file <tls_ca_file>

CA cert(s) file for TLS (in PEM format)

--tls-cert <tls_cert>

certificate file for TLS (in PEM format)

--tls-key <tls_key>

private key file for TLS (in PEM format)

--worker-port <worker_port>

Serving computation port, defaults to random

--nanny-port <nanny_port>

Serving nanny port, defaults to random

--bokeh-port <bokeh_port>

Deprecated. See –dashboard-address

--dashboard-address <dashboard_address>

Address on which to listen for diagnostics dashboard

--bokeh, --no-bokeh

Launch Bokeh Web UI [default: True]

--listen-address <listen_address>

The address to which the worker binds. Example: tcp://0.0.0.0:9000

--contact-address <contact_address>

The address the worker advertises to the scheduler for communication with it and other workers. Example: tcp://127.0.0.1:9000

--host <host>

Serving host. Should be an ip address that is visible to the scheduler and other workers. See –listen-address and –contact-address if you need different listen and contact addresses. See –interface.

--interface <interface>

Network interface like ‘eth0’ or ‘ib0’

--nthreads <nthreads>

Number of threads per process.

--nprocs <nprocs>

Number of worker processes to launch. Defaults to one.

--name <name>

A unique name for this worker like ‘worker-1’. If used with –nprocs then the process number will be appended like name-0, name-1, name-2, …

--memory-limit <memory_limit>

Bytes of memory per process that the worker can use. This can be an integer (bytes), float (fraction of total system memory), string (like 5GB or 5000M), ‘auto’, or zero for no memory management

--reconnect, --no-reconnect

Reconnect to scheduler if disconnected

--nanny, --no-nanny

Start workers in nanny process for management

--pid-file <pid_file>

File to write the process PID

--local-directory <local_directory>

Directory to place worker files

--resources <resources>

Resources for task constraints like “GPU=2 MEM=10e9”. Resources are applied separately to each worker process (only relevant when starting multiple worker processes with ‘–nprocs’).

--scheduler-file <scheduler_file>

Filename to JSON encoded scheduler information. Use with dask-scheduler –scheduler-file

--death-timeout <death_timeout>

Seconds to wait for a scheduler before closing

--bokeh-prefix <bokeh_prefix>

Prefix for the bokeh app

--preload <preload>

Module that should be loaded by each worker process like “foo.bar” or “/path/to/foo.py”

Arguments

SCHEDULER

Optional argument

PRELOAD_ARGV

Optional argument(s)

dask-ssh

Launch a distributed cluster over SSH. A ‘dask-scheduler’ process will run on the first host specified in [HOSTNAMES] or in the hostfile (unless –scheduler is specified explicitly). One or more ‘dask-worker’ processes will be run each host in [HOSTNAMES] or in the hostfile. Use command line flags to adjust how many dask-worker process are run on each host (–nprocs) and how many cpus are used by each dask-worker process (–nthreads).

dask-ssh [OPTIONS] [HOSTNAMES]...

Options

--scheduler <scheduler>

Specify scheduler node. Defaults to first address.

--scheduler-port <scheduler_port>

Specify scheduler port number. Defaults to port 8786.

--nthreads <nthreads>

Number of threads per worker process. Defaults to number of cores divided by the number of processes per host.

--nprocs <nprocs>

Number of worker processes per host. Defaults to one.

--hostfile <hostfile>

Textfile with hostnames/IP addresses

--ssh-username <ssh_username>

Username to use when establishing SSH connections.

--ssh-port <ssh_port>

Port to use for SSH connections.

--ssh-private-key <ssh_private_key>

Private key file to use for SSH connections.

--nohost

Do not pass the hostname to the worker.

--log-directory <log_directory>

Directory to use on all cluster nodes for the output of dask-scheduler and dask-worker commands.

--remote-python <remote_python>

Path to Python on remote nodes.

--memory-limit <memory_limit>

Bytes of memory that the worker can use. This can be an integer (bytes), float (fraction of total system memory), string (like 5GB or 5000M), ‘auto’, or zero for no memory management

--worker-port <worker_port>

Serving computation port, defaults to random

--nanny-port <nanny_port>

Serving nanny port, defaults to random

--remote-dask-worker <remote_dask_worker>

Worker to run. Defaults to distributed.cli.dask_worker

Arguments

HOSTNAMES

Optional argument(s)

dask-submit

dask-submit [OPTIONS] REMOTE_CLIENT_ADDRESS FILEPATH

Arguments

REMOTE_CLIENT_ADDRESS

Required argument

FILEPATH

Required argument

dask-remote

dask-remote [OPTIONS]

Options

--host <host>

IP or hostname of this server

--port <port>

Remote Client Port