Local Cluster

For convenience you can start a local cluster from your Python session.

>>> from distributed import Executor, LocalCluster
>>> c = LocalCluster()
LocalCluster("127.0.0.1:8786", workers=8, ncores=8)
>>> e = Executor(c)
<Executor: scheduler=127.0.0.1:8786 processes=8 cores=8>

Alternatively, a LocalCluster is made for you automatically if you create an Executor with no arguments.

>>> from distributed import Executor
>>> e = Executor()
>>> e
<Executor: scheduler=127.0.0.1:8786 processes=8 cores=8>
class distributed.deploy.local.LocalCluster(n_workers=None, threads_per_worker=None, nanny=True, loop=None, start=True, scheduler_port=8786, silence_logs=50, diagnostics_port=8787, services={'http': <function HTTPScheduler at 0x7f6e0849b048>}, **kwargs)[source]

Create local Scheduler and Workers

This creates a “cluster” of a scheduler and workers running on the local machine.

Parameters:

n_workers: int

Number of workers to start

threads_per_worker: int

Number of threads per each worker

nanny: boolean

If true start the workers in separate processes managed by a nanny. If False keep the workers in the main calling process

scheduler_port: int

Port of the scheduler. 8786 by default, use 0 to choose a random port

Examples

>>> c = LocalCluster()  # Create a local cluster with as many workers as cores  
>>> c  
LocalCluster("127.0.0.1:8786", workers=8, ncores=8)
>>> e = Executor(c)  # connect to local cluster  

Add a new worker to the cluster >>> w = c.start_worker(ncores=2) # doctest: +SKIP

Shut down the extra worker >>> c.remove_worker(w) # doctest: +SKIP

Start a diagnostic web server and open a new browser tab >>> c.start_diagnostics_server(show=True) # doctest: +SKIP

close()[source]

Close the cluster

start_diagnostics_server(port=8787, show=False, silence=50)[source]

Start Diagnostics Web Server

This starts a web application to show diagnostics of what is happening on the cluster. This application runs in a separate process and is generally available at the following location:

start_worker(port=0, ncores=0, **kwargs)[source]

Add a new worker to the running cluster

Parameters:

port: int (optional)

Port on which to serve the worker, defaults to 0 or random

ncores: int (optional)

Number of threads to use. Defaults to number of logical cores

nanny: boolean

If true start worker in separate process managed by a nanny

Returns:

The created Worker or Nanny object. Can be discarded.

Examples

>>> c = LocalCluster()  
>>> c.start_worker(ncores=2)  
stop_worker(w)[source]

Stop a running worker

Examples

>>> c = LocalCluster()  
>>> w = c.start_worker(ncores=2)  
>>> c.stop_worker(w)