API

Client

Future

Client Coordination

Other

Asynchronous methods

Most methods and functions can be used equally well within a blocking orasynchronous environment using Tornado coroutines. If used within a TornadoIOLoop then you should yield or await otherwise blocking operationsappropriately.

You must tell the client that you intend to use it within an asynchronousenvironment by passing the asynchronous=True keyword

# blocking
client = Client()
future = client.submit(func, *args)  # immediate, no blocking/async difference
result = client.gather(future)  # blocking
 
# asynchronous Python 2/3
client = yield Client(asynchronous=True)
future = client.submit(func, *args)  # immediate, no blocking/async difference
result = yield client.gather(future)  # non-blocking/asynchronous
 
# asynchronous Python 3
client = await Client(asynchronous=True)
future = client.submit(func, *args)  # immediate, no blocking/async difference
result = await client.gather(future)  # non-blocking/asynchronous

The asynchronous variants must be run within a Tornado coroutine. See theAsynchronous documentation for more information.

Client

• class distributed.Client(address=None, loop=None, timeout='no_default', set_as_default=True, scheduler_file=None, security=None, asynchronous=False, name=None, heartbeat_interval=None, serializers=None, deserializers=None, extensions=[], direct_to_workers=None, **kwargs)[source]
• Connect to and submit computation to a Dask cluster

The Client connects users to a Dask cluster. It provides an asynchronoususer interface around functions and futures. This class resemblesexecutors in concurrent.futures but also allows Future objectswithin submit/map calls. When a Client is instantiated it takes overall dask.compute and dask.persist calls by default.

It is also common to create a Client without specifying the scheduleraddress , like Client(). In this case the Client creates aLocalCluster in the background and connects to that. Any extrakeywords are passed from Client to LocalCluster in this case. See theLocalCluster documentation for more information.

Parameters:

• address: string, or Cluster

• This can be the address of a Scheduler server like a string'127.0.0.1:8786' or a cluster object like LocalCluster()

• timeout: int

• Timeout duration for initial connection to the scheduler

• set_as_default: bool (True)

• Claim this scheduler as the global dask scheduler

• scheduler_file: string (optional)

• Path to a file with scheduler information if available

• security: Security or bool, optional

• Optional security information. If creating a local cluster can alsopass in True, in which case temporary self-signed credentials willbe created automatically.

• asynchronous: bool (False by default)

• Set to True if using this client within async/await functions or withinTornado gen.coroutines. Otherwise this should remain False for normaluse.

• name: string (optional)

• Gives the client a name that will be included in logs generated onthe scheduler for matters relating to this client

• direct_to_workers: bool (optional)

• Whether or not to connect directly to the workers, or to askthe scheduler to serve as intermediary.

• heartbeat_interval: int Time in milliseconds between heartbeats to scheduler

• **kwargs: If you do not pass a scheduler address, Client will create aLocalCluster object, passing any extra keyword arguments.

See also

• distributed.scheduler.Scheduler
• Internal scheduler
• distributed.deploy.local.LocalCluster

Examples

Provide cluster’s scheduler node address on initialization:

>>> client = Client('127.0.0.1:8786')  # doctest: +SKIP

Use submit method to send individual computations to the cluster

>>> a = client.submit(add, 1, 2)  # doctest: +SKIP
>>> b = client.submit(add, 10, 20)  # doctest: +SKIP

Continue using submit or map on results to build up larger computations

>>> c = client.submit(add, a, b)  # doctest: +SKIP

Gather results with the gather method.

>>> client.gather(c)  # doctest: +SKIP
33

You can also call Client with no arguments in order to create your ownlocal cluster.

>>> client = Client()  # makes your own local "cluster" # doctest: +SKIP

Extra keywords will be passed directly to LocalCluster

>>> client = Client(processes=False, threads_per_worker=1)  # doctest: +SKIP

• asynchronous
• Are we running in the event loop?

This is true if the user signaled that we might be when creating theclient as in the following:

client = Client(asynchronous=True)

However, we override this expectation if we can definitively tell thatwe are running from a thread that is not the event loop. This iscommon when calling get_client() from within a worker task. Eventhough the client was originally created in asynchronous mode we mayfind ourselves in contexts when it is better to operate synchronously.

• callstack(_self, futures=None, keys=None)[source]
• The actively running call stack of all relevant keys

You can specify data of interest either by providing futures orcollections in the futures= keyword or a list of explicit keys inthe keys= keyword. If neither are provided then all call stackswill be returned.

Parameters:

- **futures: list (optional)**
- 

List of futures, defaults to all data

- **keys: list (optional)**
- 

List of key names, defaults to all data

Examples

>>> df = dd.read_parquet(...).persist()  # doctest: +SKIP
>>> client.call_stack(df)  # call on collections

>>> client.call_stack()  # Or call with no arguments for all activity  # doctest: +SKIP

• cancel(self, futures, asynchronous=None, force=False)[source]
• Cancel running futures

This stops future tasks from being scheduled if they have not yet runand deletes them if they have already run. After calling, this resultand all dependent results will no longer be accessible

Parameters:

- **futures: list of Futures**
- 
- **force: boolean (False)**
- 

Cancel this future even if other clients desire it

• close(self, timeout='no_default')[source]
• Close this client

Clients will also close automatically when your Python session ends

If you started a client without arguments like Client() then thiswill also close the local cluster that was started at the same time.

See also

- [<code>Client.restart</code>](#distributed.Client.restart)
- 

• compute(self, collections, sync=False, optimize_graph=True, workers=None, allow_other_workers=False, resources=None, retries=0, priority=0, fifo_timeout='60s', actors=None, traverse=True, **kwargs)[source]
• Compute dask collections on cluster

Parameters:

- **collections: iterable of dask objects or single dask object**
- 

Collections like dask.array or dataframe or dask.value objects

- **sync: bool (optional)**
- 

Returns Futures if False (default) or concrete values if True

- **optimize_graph: bool**
- 

Whether or not to optimize the underlying graphs

- **workers: str, list, dict**
- 

Which workers can run which parts of the computationIf a string a list then the output collections will run on the listedworkers, but other sub-computations can run anywhereIf a dict then keys should be (tuples of) collections and valuesshould be addresses or lists.

- **allow_other_workers: bool, list**
- 

If True then all restrictions in workers= are considered looseIf a list then only the keys for the listed collections are loose

- **retries: int (default to 0)**
- 

Number of allowed automatic retries if computing a result fails

- **priority: Number**
- 

Optional prioritization of task. Zero is default.Higher priorities take precedence

- **fifo_timeout: timedelta str (defaults to ’60s’)**
- 

Allowed amount of time between calls to consider the same priority

- **traverse: bool (defaults to True)**
- 

By default dask traverses builtin python collections looking fordask objects passed to compute. For large collections this canbe expensive. If none of the arguments contain any dask objects,set traverse=False to avoid doing this traversal.

- **resources: dict (defaults to {})**
- 

Defines the resources these tasks require on the worker. Canspecify global resources ({'GPU': 2}), or per-task resources({'x': {'GPU': 1}, 'y': {'SSD': 4}}), but not both.See worker resources for details on definingresources.

- **actors: bool or dict (default None)**
- 

Whether these tasks should exist on the worker as stateful actors.Specified on a global (True/False) or per-task ({'x': True, 'y': False}) basis. See Actors for additional details.

- ****kwargs:**
- 

Options to pass to the graph optimize calls Returns:

- List of Futures if input is a sequence, or a single future otherwise
- 

See also

- [<code>Client.get</code>](#distributed.Client.get)
- Normal synchronous dask.get function

Examples

>>> from dask import delayed
>>> from operator import add
>>> x = delayed(add)(1, 2)
>>> y = delayed(add)(x, x)
>>> xx, yy = client.compute([x, y])  # doctest: +SKIP
>>> xx  # doctest: +SKIP
<Future: status: finished, key: add-8f6e709446674bad78ea8aeecfee188e>
>>> xx.result()  # doctest: +SKIP
3
>>> yy.result()  # doctest: +SKIP
6

Also support single arguments

>>> xx = client.compute(x)  # doctest: +SKIP

• classmethod current()[source]

• Return global client if one exists, otherwise raise ValueError

• gather(self, futures, errors='raise', direct=None, asynchronous=None)[source]

• Gather futures from distributed memory

Accepts a future, nested container of futures, iterator, or queue.The return type will match the input type.

Parameters:

- **futures: Collection of futures**
- 

This can be a possibly nested collection of Future objects.Collections can be lists, sets, or dictionaries

- **errors: string**
- 

Either ‘raise’ or ‘skip’ if we should raise if a future has erredor skip its inclusion in the output collection

- **direct: boolean**
- 

Whether or not to connect directly to the workers, or to askthe scheduler to serve as intermediary. This can also be set whencreating the Client. Returns:

- results: a collection of the same type as the input, but now with
- 
- gathered results rather than futures
- 

See also

- [<code>Client.scatter</code>](#distributed.Client.scatter)
- Send data out to cluster

Examples

>>> from operator import add  # doctest: +SKIP
>>> c = Client('127.0.0.1:8787')  # doctest: +SKIP
>>> x = c.submit(add, 1, 2)  # doctest: +SKIP
>>> c.gather(x)  # doctest: +SKIP
3
>>> c.gather([x, [x], x])  # support lists and dicts # doctest: +SKIP
[3, [3], 3]

• get(self, dsk, keys, restrictions=None, loose_restrictions=None, resources=None, sync=True, asynchronous=None, direct=None, retries=None, priority=0, fifo_timeout='60s', actors=None, **kwargs)[source]
• Compute dask graph

Parameters:

- **dsk: dict**
- 
- **keys: object, or nested lists of objects**
- 
- **restrictions: dict (optional)**
- 

A mapping of {key: {set of worker hostnames}} that restricts wherejobs can take place

- **retries: int (default to 0)**
- 

Number of allowed automatic retries if computing a result fails

- **priority: Number**
- 

Optional prioritization of task. Zero is default.Higher priorities take precedence

- **sync: bool (optional)**
- 

Returns Futures if False or concrete values if True (default).

- **direct: bool**
- 

Whether or not to connect directly to the workers, or to askthe scheduler to serve as intermediary. This can also be set whencreating the Client.

See also

- [<code>Client.compute</code>](#distributed.Client.compute)
- Compute asynchronous collections

Examples

>>> from operator import add  # doctest: +SKIP
>>> c = Client('127.0.0.1:8787')  # doctest: +SKIP
>>> c.get({'x': (add, 1, 2)}, 'x')  # doctest: +SKIP
3

• getdataset(_self, name, **kwargs)[source]
• Get named dataset from the scheduler

See also

- [<code>Client.publish_dataset</code>](#distributed.Client.publish_dataset)
- 
- [<code>Client.list_datasets</code>](#distributed.Client.list_datasets)
- 

• getexecutor(_self, **kwargs)[source]
• Return a concurrent.futures Executor for submitting tasks on this Client

Parameters:

- ****kwargs:**
- 

Any submit()- or map()- compatible arguments, such asworkers or resources. Returns:

- An Executor object that’s fully compatible with the concurrent.futures
- 
- API.
- 

• getmetadata(_self, keys, default='no_default')[source]
• Get arbitrary metadata from scheduler

See set_metadata for the full docstring with examples

Parameters:

- **keys: key or list**
- 

Key to access. If a list then gets within a nested collection

- **default: optional**
- 

If the key does not exist then return this value instead.If not provided then this raises a KeyError if the key is notpresent

See also

- [<code>Client.set_metadata</code>](#distributed.Client.set_metadata)
- 

• classmethod getrestrictions(_collections, workers, allow_other_workers)[source]

• Get restrictions from inputs to compute/persist

• getscheduler_logs(_self, n=None)[source]

• Get logs from scheduler

Parameters:

- **n**:int
- 

Number of logs to retrive. Maxes out at 10000 by default,confiruable in config.yaml::log-length Returns:

- Logs in reversed order (newest first)
- 

• gettask_stream(_self, start=None, stop=None, count=None, plot=False, filename='task-stream.html')[source]
• Get task stream data from scheduler

This collects the data present in the diagnostic “Task Stream” plot onthe dashboard. It includes the start, stop, transfer, anddeserialization time of every task for a particular duration.

Note that the task stream diagnostic does not run by default. You maywish to call this function once before you start work to ensure thatthings start recording, and then again after you have completed.

Parameters:

- **start: Number or string**
- 

When you want to start recordingIf a number it should be the result of calling time()If a string then it should be a time difference before now,like ’60s’ or ‘500 ms’

- **stop: Number or string**
- 

When you want to stop recording

- **count: int**
- 

The number of desired records, ignored if both start and stop arespecified

- **plot: boolean, str**
- 

If true then also return a Bokeh figureIf plot == ‘save’ then save the figure to a file

- **filename: str (optional)**
- 

The filename to save to if you set plot='save' Returns:

- L: List[Dict]
- 

See also

- [<code>get_task_stream</code>](#distributed.get_task_stream)
- a context manager version of this method

Examples

>>> client.get_task_stream()  # prime plugin if not already connected
>>> x.compute()  # do some work
>>> client.get_task_stream()
[{'task': ...,
  'type': ...,
  'thread': ...,
  ...}]

Pass the plot=True or plot='save' keywords to get back a Bokehfigure

>>> data, figure = client.get_task_stream(plot='save', filename='myfile.html')

Alternatively consider the context manager

>>> from dask.distributed import get_task_stream
>>> with get_task_stream() as ts:
...     x.compute()
>>> ts.data
[...]

• getversions(_self, check=False, packages=[])[source]
• Return version info for the scheduler, all workers and myself

Parameters:

- **check**:boolean, default False
- 

raise ValueError if all required & optional packagesdo not match

- **packages**:List[str]
- 

Extra package names to check

Examples

>>> c.get_versions()  # doctest: +SKIP

>>> c.get_versions(packages=['sklearn', 'geopandas'])  # doctest: +SKIP

• getworker_logs(_self, n=None, workers=None, nanny=False)[source]
• Get logs from workers

Parameters:

- **n**:int
- 

Number of logs to retrive. Maxes out at 10000 by default,confiruable in config.yaml::log-length

- **workers**:iterable
- 

List of worker addresses to retrieve. Gets all workers by default.

- **nanny**:bool, default False
- 

Whether to get the logs from the workers (False) or the nannies (True). Ifspecified, the addresses in workers should still be the worker addresses,not the nanny addresses. Returns:

- Dictionary mapping worker address to logs.
- 
- Logs are returned in reversed order (newest first)
- 

• haswhat(_self, workers=None, **kwargs)[source]
• Which keys are held by which workers

This returns the keys of the data that are held in each worker’smemory.

Parameters:

- **workers: list (optional)**
- 

A list of worker addresses, defaults to all

See also

- [<code>Client.who_has</code>](#distributed.Client.who_has)
- 
- [<code>Client.nthreads</code>](#distributed.Client.nthreads)
- 
- [<code>Client.processing</code>](#distributed.Client.processing)
- 

Examples

>>> x, y, z = c.map(inc, [1, 2, 3])  # doctest: +SKIP
>>> wait([x, y, z])  # doctest: +SKIP
>>> c.has_what()  # doctest: +SKIP
{'192.168.1.141:46784': ['inc-1c8dd6be1c21646c71f76c16d09304ea',
                         'inc-fd65c238a7ea60f6a01bf4c8a5fcf44b',
                         'inc-1e297fc27658d7b67b3a758f16bcf47a']}

• listdatasets(_self, **kwargs)[source]
• List named datasets available on the scheduler

See also

- [<code>Client.publish_dataset</code>](#distributed.Client.publish_dataset)
- 
- [<code>Client.get_dataset</code>](#distributed.Client.get_dataset)
- 

• map(self, func, *iterables, key=None, workers=None, retries=None, resources=None, priority=0, allow_other_workers=False, fifo_timeout='100 ms', actor=False, actors=False, pure=None, **kwargs)[source]
• Map a function on a sequence of arguments

Arguments can be normal objects or Futures

Parameters:

- **func: callable**
- 
- **iterables: Iterables**
- 

List-like objects to map over. They should have the same length.

- **key: str, list**
- 

Prefix for task names if string. Explicit names if list.

- **pure: bool (defaults to True)**
- 

Whether or not the function is pure. Set pure=False forimpure functions like np.random.random.

- **workers: set, iterable of sets**
- 

A set of worker hostnames on which computations may be performed.Leave empty to default to all workers (common case)

- **allow_other_workers: bool (defaults to False)**
- 

Used with workers. Indicates whether or not the computationsmay be performed on workers that are not in the workers set(s).

- **retries: int (default to 0)**
- 

Number of allowed automatic retries if a task fails

- **priority: Number**
- 

Optional prioritization of task. Zero is default.Higher priorities take precedence

- **fifo_timeout: str timedelta (default ‘100ms’)**
- 

Allowed amount of time between calls to consider the same priority

- **resources: dict (defaults to {})**
- 

Defines the resources each instance of this mapped task requireson the worker; e.g. {'GPU': 2}. Seeworker resources for details on definingresources.

- **actor: bool (default False)**
- 

Whether these tasks should exist on the worker as stateful actors.See Actors for additional details.

- **actors: bool (default False)**
- 

Alias for actor

- ****kwargs: dict**
- 

Extra keywords to send to the function.Large values will be included explicitly in the task graph. Returns:

- List, iterator, or Queue of futures, depending on the type of the
- 
- inputs.
- 

See also

- [<code>Client.submit</code>](#distributed.Client.submit)
- Submit a single function

Examples

>>> L = client.map(func, sequence)  # doctest: +SKIP

• nbytes(self, keys=None, summary=True, **kwargs)[source]
• The bytes taken up by each key on the cluster

This is as measured by sys.getsizeof which may not accuratelyreflect the true cost.

Parameters:

- **keys: list (optional)**
- 

A list of keys, defaults to all keys

- **summary: boolean, (optional)**
- 

Summarize keys into key types

See also

- [<code>Client.who_has</code>](#distributed.Client.who_has)
- 

Examples

>>> x, y, z = c.map(inc, [1, 2, 3])  # doctest: +SKIP
>>> c.nbytes(summary=False)  # doctest: +SKIP
{'inc-1c8dd6be1c21646c71f76c16d09304ea': 28,
 'inc-1e297fc27658d7b67b3a758f16bcf47a': 28,
 'inc-fd65c238a7ea60f6a01bf4c8a5fcf44b': 28}

>>> c.nbytes(summary=True)  # doctest: +SKIP
{'inc': 84}

• ncores(self, workers=None, **kwargs)
• The number of threads/cores available on each worker node

Parameters:

- **workers: list (optional)**
- 

A list of workers that we care about specifically.Leave empty to receive information about all workers.

See also

- [<code>Client.who_has</code>](#distributed.Client.who_has)
- 
- [<code>Client.has_what</code>](#distributed.Client.has_what)
- 

Examples

>>> c.threads()  # doctest: +SKIP
{'192.168.1.141:46784': 8,
 '192.167.1.142:47548': 8,
 '192.167.1.143:47329': 8,
 '192.167.1.144:37297': 8}

• normalizecollection(_self, collection)[source]
• Replace collection’s tasks by already existing futures if they exist

This normalizes the tasks within a collections task graph against theknown futures within the scheduler. It returns a copy of thecollection with a task graph that includes the overlapping futures.

See also

- [<code>Client.persist</code>](#distributed.Client.persist)
- trigger computation of collection’s tasks

Examples

>>> len(x.__dask_graph__())  # x is a dask collection with 100 tasks  # doctest: +SKIP
100
>>> set(client.futures).intersection(x.__dask_graph__())  # some overlap exists  # doctest: +SKIP
10

>>> x = client.normalize_collection(x)  # doctest: +SKIP
>>> len(x.__dask_graph__())  # smaller computational graph  # doctest: +SKIP
20

• nthreads(self, workers=None, **kwargs)[source]
• The number of threads/cores available on each worker node

Parameters:

- **workers: list (optional)**
- 

A list of workers that we care about specifically.Leave empty to receive information about all workers.

See also

- [<code>Client.who_has</code>](#distributed.Client.who_has)
- 
- [<code>Client.has_what</code>](#distributed.Client.has_what)
- 

Examples

>>> c.threads()  # doctest: +SKIP
{'192.168.1.141:46784': 8,
 '192.167.1.142:47548': 8,
 '192.167.1.143:47329': 8,
 '192.167.1.144:37297': 8}

• persist(self, collections, optimize_graph=True, workers=None, allow_other_workers=None, resources=None, retries=None, priority=0, fifo_timeout='60s', actors=None, **kwargs)[source]
• Persist dask collections on cluster

Starts computation of the collection on the cluster in the background.Provides a new dask collection that is semantically identical to theprevious one, but now based off of futures currently in execution.

Parameters:

- **collections: sequence or single dask object**
- 

Collections like dask.array or dataframe or dask.value objects

- **optimize_graph: bool**
- 

Whether or not to optimize the underlying graphs

- **workers: str, list, dict**
- 

Which workers can run which parts of the computationIf a string a list then the output collections will run on the listedworkers, but other sub-computations can run anywhereIf a dict then keys should be (tuples of) collections and valuesshould be addresses or lists.

- **allow_other_workers: bool, list**
- 

If True then all restrictions in workers= are considered looseIf a list then only the keys for the listed collections are loose

- **retries: int (default to 0)**
- 

Number of allowed automatic retries if computing a result fails

- **priority: Number**
- 

Optional prioritization of task. Zero is default.Higher priorities take precedence

- **fifo_timeout: timedelta str (defaults to ’60s’)**
- 

Allowed amount of time between calls to consider the same priority

- **resources: dict (defaults to {})**
- 

Defines the resources these tasks require on the worker. Canspecify global resources ({'GPU': 2}), or per-task resources({'x': {'GPU': 1}, 'y': {'SSD': 4}}), but not both.See worker resources for details on definingresources.

- **actors: bool or dict (default None)**
- 

Whether these tasks should exist on the worker as stateful actors.Specified on a global (True/False) or per-task ({'x': True, 'y': False}) basis. See Actors for additional details.

- ****kwargs:**
- 

Options to pass to the graph optimize calls Returns:

- List of collections, or single collection, depending on type of input.
- 

See also

- [<code>Client.compute</code>](#distributed.Client.compute)
- 

Examples

>>> xx = client.persist(x)  # doctest: +SKIP
>>> xx, yy = client.persist([x, y])  # doctest: +SKIP

• processing(self, workers=None)[source]
• The tasks currently running on each worker

Parameters:

- **workers: list (optional)**
- 

A list of worker addresses, defaults to all

See also

- [<code>Client.who_has</code>](#distributed.Client.who_has)
- 
- [<code>Client.has_what</code>](#distributed.Client.has_what)
- 
- [<code>Client.nthreads</code>](#distributed.Client.nthreads)
- 

Examples

>>> x, y, z = c.map(inc, [1, 2, 3])  # doctest: +SKIP
>>> c.processing()  # doctest: +SKIP
{'192.168.1.141:46784': ['inc-1c8dd6be1c21646c71f76c16d09304ea',
                         'inc-fd65c238a7ea60f6a01bf4c8a5fcf44b',
                         'inc-1e297fc27658d7b67b3a758f16bcf47a']}

• profile(self, key=None, start=None, stop=None, workers=None, merge_workers=True, plot=False, filename=None, server=False, scheduler=False)[source]
• Collect statistical profiling information about recent work

Parameters:

- **key: str**
- 

Key prefix to select, this is typically a function name like ‘inc’Leave as None to collect all data

- **start: time**
- 
- **stop: time**
- 
- **workers: list**
- 

List of workers to restrict profile information

- **server**:bool
- 

If true, return the profile of the worker’s administrative threadrather than the worker threads.This is useful when profiling Dask itself, rather than user code.

- **scheduler: bool**
- 

If true, return the profile information from the scheduler’sadministrative thread rather than the workers.This is useful when profiling Dask’s scheduling itself.

- **plot: boolean or string**
- 

Whether or not to return a plot object

- **filename: str**
- 

Filename to save the plot

Examples

>>> client.profile()  # call on collections
>>> client.profile(filename='dask-profile.html')  # save to html file

• publishdataset(_self, *args, **kwargs)[source]
• Publish named datasets to scheduler

This stores a named reference to a dask collection or list of futureson the scheduler. These references are available to other Clientswhich can download the collection or futures with get_dataset.

Datasets are not immediately computed. You may wish to callClient.persist prior to publishing a dataset.

Parameters:

- **args**:list of objects to publish as name
- 
- **name**:optional name of the dataset to publish
- 
- **kwargs: dict**
- 

named collections to publish on the scheduler Returns:

- None
- 

See also

- [<code>Client.list_datasets</code>](#distributed.Client.list_datasets)
- 
- [<code>Client.get_dataset</code>](#distributed.Client.get_dataset)
- 
- [<code>Client.unpublish_dataset</code>](#distributed.Client.unpublish_dataset)
- 
- [<code>Client.persist</code>](#distributed.Client.persist)
- 

Examples

Publishing client:

>>> df = dd.read_csv('s3://...')  # doctest: +SKIP
>>> df = c.persist(df) # doctest: +SKIP
>>> c.publish_dataset(my_dataset=df)  # doctest: +SKIP

Alternative invocation>>> c.publish_dataset(df, name=’my_dataset’)

Receiving client:

>>> c.list_datasets()  # doctest: +SKIP
['my_dataset']
>>> df2 = c.get_dataset('my_dataset')  # doctest: +SKIP

• rebalance(self, futures=None, workers=None, **kwargs)[source]
• Rebalance data within network

Move data between workers to roughly balance memory burden. Thiseither affects a subset of the keys/workers or the entire network,depending on keyword arguments.

This operation is generally not well tested against normal operation ofthe scheduler. It it not recommended to use it while waiting oncomputations.

Parameters:

- **futures: list, optional**
- 

A list of futures to balance, defaults all data

- **workers: list, optional**
- 

A list of workers on which to balance, defaults to all workers

• registerworker_callbacks(_self, setup=None)[source]
• Registers a setup callback function for all current and future workers.

This registers a new setup function for workers in this cluster. Thefunction will run immediately on all currently connected workers. Itwill also be run upon connection by any workers that are added in thefuture. Multiple setup functions can be registered - these will becalled in the order they were added.

If the function takes an input argument named dask_worker thenthat variable will be populated with the worker itself.

Parameters:

- **setup**:callable(dask_worker: Worker) -> None
- 

Function to register and run on all workers

• registerworker_plugin(_self, plugin=None, name=None)[source]
• Registers a lifecycle worker plugin for all current and future workers.

This registers a new object to handle setup, task state transitions andteardown for workers in this cluster. The plugin will instantiate itselfon all currently connected workers. It will also be run on any workerthat connects in the future.

The plugin may include methods setup, teardown, andtransition. See the dask.distributed.WorkerPlugin class or theexamples below for the interface and docstrings. It must beserializable with the pickle or cloudpickle modules.

If the plugin has a name attribute, or if the name= keyword isused then that will control idempotency. A a plugin with that name hasalready registered then any future plugins will not run.

For alternatives to plugins, you may also wish to look into preloadscripts.

Parameters:

- **plugin: WorkerPlugin**
- 

The plugin object to pass to the workers

- **name: str, optional**
- 

A name for the plugin.Registering a plugin with the same name will have no effect.

See also

- <code>distributed.WorkerPlugin</code>
- 

Examples

>>> class MyPlugin(WorkerPlugin):
...     def __init__(self, *args, **kwargs):
...         pass  # the constructor is up to you
...     def setup(self, worker: dask.distributed.Worker):
...         pass
...     def teardown(self, worker: dask.distributed.Worker):
...         pass
...     def transition(self, key: str, start: str, finish: str, **kwargs):
...         pass

>>> plugin = MyPlugin(1, 2, 3)
>>> client.register_worker_plugin(plugin)

You can get access to the plugin with the get_worker function

>>> client.register_worker_plugin(other_plugin, name='my-plugin')
>>> def f():
...    worker = get_worker()
...    plugin = worker.plugins['my-plugin']
...    return plugin.my_state

>>> future = client.run(f)

• replicate(self, futures, n=None, workers=None, branching_factor=2, **kwargs)[source]
• Set replication of futures within network

Copy data onto many workers. This helps to broadcast frequentlyaccessed data and it helps to improve resilience.

This performs a tree copy of the data throughout the networkindividually on each piece of data. This operation blocks untilcomplete. It does not guarantee replication of data to future workers.

Parameters:

- **futures: list of futures**
- 

Futures we wish to replicate

- **n: int, optional**
- 

Number of processes on the cluster on which to replicate the data.Defaults to all.

- **workers: list of worker addresses**
- 

Workers on which we want to restrict the replication.Defaults to all.

- **branching_factor: int, optional**
- 

The number of workers that can copy data in each generation

See also

- [<code>Client.rebalance</code>](#distributed.Client.rebalance)
- 

Examples

>>> x = c.submit(func, *args)  # doctest: +SKIP
>>> c.replicate([x])  # send to all workers  # doctest: +SKIP
>>> c.replicate([x], n=3)  # send to three workers  # doctest: +SKIP
>>> c.replicate([x], workers=['alice', 'bob'])  # send to specific  # doctest: +SKIP
>>> c.replicate([x], n=1, workers=['alice', 'bob'])  # send to one of specific workers  # doctest: +SKIP
>>> c.replicate([x], n=1)  # reduce replications # doctest: +SKIP

• restart(self, **kwargs)[source]
• Restart the distributed network

This kills all active work, deletes all data on the network, andrestarts the worker processes.

• retireworkers(_self, workers=None, close_workers=True, **kwargs)[source]
• Retire certain workers on the scheduler

See dask.distributed.Scheduler.retire_workers for the full docstring.

See also

- <code>dask.distributed.Scheduler.retire_workers</code>
- 

Examples

You can get information about active workers using the following:>>> workers = client.scheduler_info()[‘workers’]

From that list you may want to select some workers to close>>> client.retire_workers(workers=[‘tcp://address:port’, …])

• retry(self, futures, asynchronous=None)[source]
• Retry failed futures

Parameters:

- **futures: list of Futures**
- 

• run(self, function, *args, **kwargs)[source]
• Run a function on all workers outside of task scheduling system

This calls a function on all currently known workers immediately,blocks until those results come back, and returns the resultsasynchronously as a dictionary keyed by worker address. This methodif generally used for side effects, such and collecting diagnosticinformation or installing libraries.

If your function takes an input argument named dask_worker thenthat variable will be populated with the worker itself.

Parameters:

- **function: callable**
- 
- ***args: arguments for remote function**
- 
- ****kwargs: keyword arguments for remote function**
- 
- **workers: list**
- 

Workers on which to run the function. Defaults to all known workers.

- **wait: boolean (optional)**
- 

If the function is asynchronous whether or not to wait until thatfunction finishes.

- **nanny**:bool, defualt False
- 

Whether to run function on the nanny. By default, the functionis run on the worker process. If specified, the addresses inworkers should still be the worker addresses, not the nanny addresses.

Examples

>>> c.run(os.getpid)  # doctest: +SKIP
{'192.168.0.100:9000': 1234,
 '192.168.0.101:9000': 4321,
 '192.168.0.102:9000': 5555}

Restrict computation to particular workers with the workers=keyword argument.

>>> c.run(os.getpid, workers=['192.168.0.100:9000',
...                           '192.168.0.101:9000'])  # doctest: +SKIP
{'192.168.0.100:9000': 1234,
 '192.168.0.101:9000': 4321}

>>> def get_status(dask_worker):
...     return dask_worker.status

>>> c.run(get_hostname)  # doctest: +SKIP
{'192.168.0.100:9000': 'running',
 '192.168.0.101:9000': 'running}

Run asynchronous functions in the background:

>>> async def print_state(dask_worker):  # doctest: +SKIP
...    while True:
...        print(dask_worker.status)
...        await asyncio.sleep(1)

>>> c.run(print_state, wait=False)  # doctest: +SKIP

• runcoroutine(_self, function, *args, **kwargs)[source]
• Spawn a coroutine on all workers.

This spaws a coroutine on all currently known workers and then waitsfor the coroutine on each worker. The coroutines’ results are returnedas a dictionary keyed by worker address.

Parameters:

- **function: a coroutine function**
- 
  - (typically a function wrapped in gen.coroutine or
  - 

a Python 3.5+ async function)

- ***args: arguments for remote function**
- 
- ****kwargs: keyword arguments for remote function**
- 
- **wait: boolean (default True)**
- 

Whether to wait for coroutines to end.

- **workers: list**
- 

Workers on which to run the function. Defaults to all known workers.

• runon_scheduler(_self, function, *args, **kwargs)[source]
• Run a function on the scheduler process

This is typically used for live debugging. The function should take akeyword argument dask_scheduler=, which will be given the schedulerobject itself.

See also

- [<code>Client.run</code>](#distributed.Client.run)
- Run a function on all workers
- [<code>Client.start_ipython_scheduler</code>](#distributed.Client.start_ipython_scheduler)
- Start an IPython session on scheduler

Examples

>>> def get_number_of_tasks(dask_scheduler=None):
...     return len(dask_scheduler.tasks)

>>> client.run_on_scheduler(get_number_of_tasks)  # doctest: +SKIP
100

Run asynchronous functions in the background:

>>> async def print_state(dask_scheduler):  # doctest: +SKIP
...    while True:
...        print(dask_scheduler.status)
...        await asyncio.sleep(1)

>>> c.run(print_state, wait=False)  # doctest: +SKIP

• scatter(self, data, workers=None, broadcast=False, direct=None, hash=True, timeout='no_default', asynchronous=None)[source]
• Scatter data into distributed memory

This moves data from the local client process into the workers of thedistributed scheduler. Note that it is often better to submit jobs toyour workers to have them load the data rather than loading datalocally and then scattering it out to them.

Parameters:

- **data: list, dict, or object**
- 

Data to scatter out to workers. Output type matches input type.

- **workers: list of tuples (optional)**
- 

Optionally constrain locations of data.Specify workers as hostname/port pairs, e.g. ('127.0.0.1', 8787).

- **broadcast: bool (defaults to False)**
- 

Whether to send each data element to all workers.By default we round-robin based on number of cores.

- **direct: bool (defaults to automatically check)**
- 

Whether or not to connect directly to the workers, or to askthe scheduler to serve as intermediary. This can also be set whencreating the Client.

- **hash: bool (optional)**
- 

Whether or not to hash data to determine key.If False then this uses a random key Returns:

- List, dict, iterator, or queue of futures matching the type of input.
- 

See also

- [<code>Client.gather</code>](#distributed.Client.gather)
- Gather data back to local process

Examples

>>> c = Client('127.0.0.1:8787')  # doctest: +SKIP
>>> c.scatter(1) # doctest: +SKIP
<Future: status: finished, key: c0a8a20f903a4915b94db8de3ea63195>

>>> c.scatter([1, 2, 3])  # doctest: +SKIP
[<Future: status: finished, key: c0a8a20f903a4915b94db8de3ea63195>,
 <Future: status: finished, key: 58e78e1b34eb49a68c65b54815d1b158>,
 <Future: status: finished, key: d3395e15f605bc35ab1bac6341a285e2>]

>>> c.scatter({'x': 1, 'y': 2, 'z': 3})  # doctest: +SKIP
{'x': <Future: status: finished, key: x>,
 'y': <Future: status: finished, key: y>,
 'z': <Future: status: finished, key: z>}

Constrain location of data to subset of workers

>>> c.scatter([1, 2, 3], workers=[('hostname', 8788)])   # doctest: +SKIP

Broadcast data to all workers

>>> [future] = c.scatter([element], broadcast=True)  # doctest: +SKIP

Send scattered data to parallelized function using client futuresinterface

>>> data = c.scatter(data, broadcast=True)  # doctest: +SKIP
>>> res = [c.submit(func, data, i) for i in range(100)]

• schedulerinfo(_self, **kwargs)[source]
• Basic information about the workers in the cluster

Examples

>>> c.scheduler_info()  # doctest: +SKIP
{'id': '2de2b6da-69ee-11e6-ab6a-e82aea155996',
 'services': {},
 'type': 'Scheduler',
 'workers': {'127.0.0.1:40575': {'active': 0,
                                 'last-seen': 1472038237.4845693,
                                 'name': '127.0.0.1:40575',
                                 'services': {},
                                 'stored': 0,
                                 'time-delay': 0.0061032772064208984}}}

• setmetadata(_self, key, value)[source]
• Set arbitrary metadata in the scheduler

This allows you to store small amounts of data on the central schedulerprocess for administrative purposes. Data should be msgpackserializable (ints, strings, lists, dicts)

If the key corresponds to a task then that key will be cleaned up whenthe task is forgotten by the scheduler.

If the key is a list then it will be assumed that you want to indexinto a nested dictionary structure using those keys. For example ifyou call the following:

>>> client.set_metadata(['a', 'b', 'c'], 123)

Then this is the same as setting

>>> scheduler.task_metadata['a']['b']['c'] = 123

The lower level dictionaries will be created on demand.

See also

- [<code>get_metadata</code>](#distributed.Client.get_metadata)
- 

Examples

>>> client.set_metadata('x', 123)  # doctest: +SKIP
>>> client.get_metadata('x')  # doctest: +SKIP
123

>>> client.set_metadata(['x', 'y'], 123)  # doctest: +SKIP
>>> client.get_metadata('x')  # doctest: +SKIP
{'y': 123}

>>> client.set_metadata(['x', 'w', 'z'], 456)  # doctest: +SKIP
>>> client.get_metadata('x')  # doctest: +SKIP
{'y': 123, 'w': {'z': 456}}

>>> client.get_metadata(['x', 'w'])  # doctest: +SKIP
{'z': 456}

• shutdown(self)[source]
• Shut down the connected scheduler and workers

Note, this may disrupt other clients that may be using the samescheudler and workers.

See also

- [<code>Client.close</code>](#distributed.Client.close)
- close only this client

• start(self, **kwargs)[source]

• Start scheduler running in separate thread

• startipython_scheduler(_self, magic_name='scheduler_if_ipython', qtconsole=False, qtconsole_args=None)[source]

• Start IPython kernel on the scheduler

Parameters:

- **magic_name: str or None (optional)**
- 

If defined, register IPython magic with this name forexecuting code on the scheduler.If not defined, register %scheduler magic if IPython is running.

- **qtconsole: bool (optional)**
- 

If True, launch a Jupyter QtConsole connected to the worker(s).

- **qtconsole_args: list(str) (optional)**
- 

Additional arguments to pass to the qtconsole on startup. Returns:

- connection_info: dict
- 

connection_info dict containing info necessaryto connect Jupyter clients to the scheduler.

See also

- [<code>Client.start_ipython_workers</code>](#distributed.Client.start_ipython_workers)
- Start IPython on the workers

Examples

>>> c.start_ipython_scheduler() # doctest: +SKIP
>>> %scheduler scheduler.processing  # doctest: +SKIP
{'127.0.0.1:3595': {'inc-1', 'inc-2'},
 '127.0.0.1:53589': {'inc-2', 'add-5'}}

>>> c.start_ipython_scheduler(qtconsole=True) # doctest: +SKIP

• startipython_workers(_self, workers=None, magic_names=False, qtconsole=False, qtconsole_args=None)[source]
• Start IPython kernels on workers

Parameters:

- **workers: list (optional)**
- 

A list of worker addresses, defaults to all

- **magic_names: str or list(str) (optional)**
- 

If defined, register IPython magics with these names forexecuting code on the workers. If string has asterix then expandasterix into 0, 1, …, n for n workers

- **qtconsole: bool (optional)**
- 

If True, launch a Jupyter QtConsole connected to the worker(s).

- **qtconsole_args: list(str) (optional)**
- 

Additional arguments to pass to the qtconsole on startup. Returns:

- iter_connection_info: list
- 

List of connection_info dicts containing info necessaryto connect Jupyter clients to the workers.

See also

- [<code>Client.start_ipython_scheduler</code>](#distributed.Client.start_ipython_scheduler)
- start ipython on the scheduler

Examples

>>> info = c.start_ipython_workers() # doctest: +SKIP
>>> %remote info['192.168.1.101:5752'] worker.data  # doctest: +SKIP
{'x': 1, 'y': 100}

>>> c.start_ipython_workers('192.168.1.101:5752', magic_names='w') # doctest: +SKIP
>>> %w worker.data  # doctest: +SKIP
{'x': 1, 'y': 100}

>>> c.start_ipython_workers('192.168.1.101:5752', qtconsole=True) # doctest: +SKIP

Add asterix * in magic names to add one magic per worker

>>> c.start_ipython_workers(magic_names='w_*') # doctest: +SKIP
>>> %w_0 worker.data  # doctest: +SKIP
{'x': 1, 'y': 100}
>>> %w_1 worker.data  # doctest: +SKIP
{'z': 5}

• submit(self, func, *args, key=None, workers=None, resources=None, retries=None, priority=0, fifo_timeout='100 ms', allow_other_workers=False, actor=False, actors=False, pure=None, **kwargs)[source]
• Submit a function application to the scheduler

Parameters:

- **func: callable**
- 
- ***args:**
- 
- ****kwargs:**
- 
- **pure: bool (defaults to True)**
- 

Whether or not the function is pure. Set pure=False forimpure functions like np.random.random.

- **workers: set, iterable of sets**
- 

A set of worker hostnames on which computations may be performed.Leave empty to default to all workers (common case)

- **key: str**
- 

Unique identifier for the task. Defaults to function-name and hash

- **allow_other_workers: bool (defaults to False)**
- 

Used with workers. Indicates whether or not the computationsmay be performed on workers that are not in the workers set(s).

- **retries: int (default to 0)**
- 

Number of allowed automatic retries if the task fails

- **priority: Number**
- 

Optional prioritization of task. Zero is default.Higher priorities take precedence

- **fifo_timeout: str timedelta (default ‘100ms’)**
- 

Allowed amount of time between calls to consider the same priority

- **resources: dict (defaults to {})**
- 

Defines the resources this job requires on the worker; e.g.{'GPU': 2}. See worker resources for detailson defining resources.

- **actor: bool (default False)**
- 

Whether this task should exist on the worker as a stateful actor.See Actors for additional details.

- **actors: bool (default False)**
- 

Alias for actor Returns:

- Future
- 

See also

- [<code>Client.map</code>](#distributed.Client.map)
- Submit on many arguments at once

Examples

>>> c = client.submit(add, a, b)  # doctest: +SKIP

• unpublishdataset(_self, name, **kwargs)[source]
• Remove named datasets from scheduler

See also

- [<code>Client.publish_dataset</code>](#distributed.Client.publish_dataset)
- 

Examples

>>> c.list_datasets()  # doctest: +SKIP
['my_dataset']
>>> c.unpublish_datasets('my_dataset')  # doctest: +SKIP
>>> c.list_datasets()  # doctest: +SKIP
[]

• uploadfile(_self, filename, **kwargs)[source]
• Upload local package to workers

This sends a local file up to all worker nodes. This file is placedinto a temporary directory on Python’s system path so any .py, .eggor .zip files will be importable.

Parameters:

- **filename: string**
- 

Filename of .py, .egg or .zip file to send to workers

Examples

>>> client.upload_file('mylibrary.egg')  # doctest: +SKIP
>>> from mylibrary import myfunc  # doctest: +SKIP
>>> L = c.map(myfunc, seq)  # doctest: +SKIP

• waitfor_workers(_self, n_workers=0)[source]

• Blocking call to wait for n workers before continuing

• whohas(_self, futures=None, **kwargs)[source]

• The workers storing each future’s data

Parameters:

- **futures: list (optional)**
- 

A list of futures, defaults to all data

See also

- [<code>Client.has_what</code>](#distributed.Client.has_what)
- 
- [<code>Client.nthreads</code>](#distributed.Client.nthreads)
- 

Examples

>>> x, y, z = c.map(inc, [1, 2, 3])  # doctest: +SKIP
>>> wait([x, y, z])  # doctest: +SKIP
>>> c.who_has()  # doctest: +SKIP
{'inc-1c8dd6be1c21646c71f76c16d09304ea': ['192.168.1.141:46784'],
 'inc-1e297fc27658d7b67b3a758f16bcf47a': ['192.168.1.141:46784'],
 'inc-fd65c238a7ea60f6a01bf4c8a5fcf44b': ['192.168.1.141:46784']}

>>> c.who_has([x, y])  # doctest: +SKIP
{'inc-1c8dd6be1c21646c71f76c16d09304ea': ['192.168.1.141:46784'],
 'inc-1e297fc27658d7b67b3a758f16bcf47a': ['192.168.1.141:46784']}

• writescheduler_file(_self, scheduler_file)[source]
• Write the scheduler information to a json file.

This facilitates easy sharing of scheduler information using a filesystem. The scheduler file can be used to instantiate a second Clientusing the same scheduler.

Parameters:

- **scheduler_file: str**
- 

Path to a write the scheduler file.

Examples

>>> client = Client()  # doctest: +SKIP
>>> client.write_scheduler_file('scheduler.json')  # doctest: +SKIP
# connect to previous client's scheduler
>>> client2 = Client(scheduler_file='scheduler.json')  # doctest: +SKIP

• class distributed.recreateexceptions.ReplayExceptionClient(_client)[source]
• A plugin for the client allowing replay of remote exceptions locally

Adds the following methods (and their async variants)to the given client:

• recreate_error_locally: main user method
• • get_futures_error: gets the task, its details and dependencies,
  • responsible for failure of the given future.
• getfutures_error(_self, future)[source]
• Ask the scheduler details of the sub-task of the given failed future

When a future evaluates to a status of “error”, i.e., an exceptionwas raised in a task within its graph, we an get information fromthe scheduler. This function gets the details of the specific taskthat raised the exception and led to the error, but does not fetchdata from the cluster or execute the function.

Parameters:

- **future**:future that failed, having <code>status==&#34;error&#34;</code>, typically
- 

after an attempt to gather() shows a stack-stace. Returns:

- Tuple:
- 
- 
  - the function that raised an exception
  - 
- 
  - argument list (a tuple), may include values and keys
  - 
- 
  - keyword arguments (a dictionary), may include values and keys
  - 
- 
  - list of keys that the function requires to be fetched to run
  - 

See also

- [<code>ReplayExceptionClient.recreate_error_locally</code>](#distributed.recreate_exceptions.ReplayExceptionClient.recreate_error_locally)
- 

• recreateerror_locally(_self, future)[source]
• For a failed calculation, perform the blamed task locally for debugging.

This operation should be performed after a future (result of gather,compute, etc) comes back with a status of “error”, if the stack-trace is not informative enough to diagnose the problem. The specifictask (part of the graph pointing to the future) responsible for theerror will be fetched from the scheduler, together with the values ofits inputs. The function will then be executed, so that pdb canbe used for debugging.

Parameters:

- **future**:future or collection that failed
- 

The same thing as was given to gather, but came back withan exception/stack-trace. Can also be a (persisted) dask collectioncontaining any errored futures. Returns:

- Nothing; the function runs and should raise an exception, allowing
- 
- the debugger to run.
- 

Examples

>>> future = c.submit(div, 1, 0)         # doctest: +SKIP
>>> future.status                        # doctest: +SKIP
'error'
>>> c.recreate_error_locally(future)     # doctest: +SKIP
ZeroDivisionError: division by zero

If you’re in IPython you might take this opportunity to use pdb

>>> %pdb                                 # doctest: +SKIP
Automatic pdb calling has been turned ON

>>> c.recreate_error_locally(future)     # doctest: +SKIP
ZeroDivisionError: division by zero
      1 def div(x, y):
----> 2     return x / y
ipdb>

Future

• class distributed.Future(key, client=None, inform=True, state=None)[source]
• A remotely running computation

A Future is a local proxy to a result running on a remote worker. A usermanages future objects in the local Python process to determine whathappens in the larger cluster.

Parameters:

• key: str, or tuple

• Key of remote data to which this future refers

• client: Client

• Client that should own this future. Defaults to _get_global_client()

• inform: bool

• Do we inform the scheduler that we need an update on this future

See also

• Client
• Creates futures

Examples

Futures typically emerge from Client computations

>>> my_future = client.submit(add, 1, 2)  # doctest: +SKIP

We can track the progress and results of a future

>>> my_future  # doctest: +SKIP
<Future: status: finished, key: add-8f6e709446674bad78ea8aeecfee188e>

We can get the result or the exception and traceback from the future

>>> my_future.result()  # doctest: +SKIP

• adddone_callback(_self, fn)[source]
• Call callback on future when callback has finished

The callback fn should take the future as its only argument. Thiswill be called regardless of if the future completes successfully,errs, or is cancelled

The callback is executed in a separate thread.

• cancel(self, **kwargs)[source]
• Cancel request to run this future

See also

- [<code>Client.cancel</code>](#distributed.Client.cancel)
- 

• cancelled(self)[source]

• Returns True if the future has been cancelled

• done(self)[source]

• Is the computation complete?

• exception(self, timeout=None, **kwargs)[source]

• Return the exception of a failed task

If timeout seconds are elapsed before returning, adask.distributed.TimeoutError is raised.

See also

- [<code>Future.traceback</code>](#distributed.Future.traceback)
- 

• result(self, timeout=None)[source]
• Wait until computation completes, gather result to local process.

If timeout seconds are elapsed before returning, adask.distributed.TimeoutError is raised.

• retry(self, **kwargs)[source]
• Retry this future if it has failed

See also

- [<code>Client.retry</code>](#distributed.Client.retry)
-