Diagnostics (local)

Profiling parallel code can be challenging, but dask.diagnostics providesfunctionality to aid in profiling and inspecting execution with thelocal task scheduler.

This page describes the following few built-in options:

• ProgressBar
• Profiler
• ResourceProfiler
• CacheProfilerFurthermore, this page then provides instructions on how to build your own custom diagnostic.

Progress Bar

The ProgressBar class builds on the scheduler callbacks described above todisplay a progress bar in the terminal or notebook during computation. This cangive a nice feedback during long running graph execution. It can be used as acontext manager around calls to get or compute to profile thecomputation:

>>> from dask.diagnostics import ProgressBar
>>> a = da.random.normal(size=(10000, 10000), chunks=(1000, 1000))
>>> res = a.dot(a.T).mean(axis=0)
 
>>> with ProgressBar():
...     out = res.compute()
[########################################] | 100% Completed | 17.1 s

or registered globally using the register method:

>>> pbar = ProgressBar()
>>> pbar.register()
>>> out = res.compute()
[########################################] | 100% Completed | 17.1 s

To unregister from the global callbacks, call the unregister method:

>>> pbar.unregister()

Profiler

Dask provides a few tools for profiling execution. As with the ProgressBar,they each can be used as context managers or registered globally.

The Profiler class is used to profile Dask’s execution at the task level.During execution, it records the following information for each task:

• Key
• Task
• Start time in seconds since the epoch
• Finish time in seconds since the epoch
• Worker id

ResourceProfiler

The ResourceProfiler class is used to profile Dask’s execution at theresource level. During execution, it records the following informationfor each timestep:

• Time in seconds since the epoch
• Memory usage in MB
• % CPU usageThe default timestep is 1 second, but can be set manually using the dtkeyword:

>>> from dask.diagnostics import ResourceProfiler
>>> rprof = ResourceProfiler(dt=0.5)

CacheProfiler

The CacheProfiler class is used to profile Dask’s execution at the schedulercache level. During execution, it records the following information for eachtask:

• Key
• Task
• Size metric
• Cache entry time in seconds since the epoch
• Cache exit time in seconds since the epochHere the size metric is the output of a function called on the result of eachtask. The default metric is to count each task (metric is 1 for all tasks).Other functions may be used as a metric instead through the metric keyword.For example, the nbytes function found in cachey can be used to measurethe number of bytes in the scheduler cache:

>>> from dask.diagnostics import CacheProfiler
>>> from cachey import nbytes
>>> cprof = CacheProfiler(metric=nbytes)

Example

As an example to demonstrate using the diagnostics, we’ll profile some linearalgebra done with Dask Array. We’ll create a random array, take its QRdecomposition, and then reconstruct the initial array by multiplying the Q andR components together. Note that since the profilers (and all diagnostics) arejust context managers, multiple profilers can be used in a with block:

>>> import dask.array as da
>>> from dask.diagnostics import Profiler, ResourceProfiler, CacheProfiler
>>> a = da.random.random(size=(10000, 1000), chunks=(1000, 1000))
>>> q, r = da.linalg.qr(a)
>>> a2 = q.dot(r)
 
>>> with Profiler() as prof, ResourceProfiler(dt=0.25) as rprof,
...         CacheProfiler() as cprof:
...     out = a2.compute()

The results of each profiler are stored in their results attribute as alist of namedtuple objects:

>>> prof.results[0]
TaskData(key=('tsqr-8d16e396b237bf7a731333130d310cb9_QR_st1', 5, 0),
         task=(qr, (_apply_random, 'random_sample', 1060164455, (1000, 1000), (), {})),
         start_time=1454368444.493292,
         end_time=1454368444.902987,
         worker_id=4466937856)
 
>>> rprof.results[0]
ResourceData(time=1454368444.078748, mem=74.100736, cpu=0.0)
 
>>> cprof.results[0]
CacheData(key=('tsqr-8d16e396b237bf7a731333130d310cb9_QR_st1', 7, 0),
          task=(qr, (_apply_random, 'random_sample', 1310656009, (1000, 1000), (), {})),
          metric=1,
          cache_time=1454368444.49662,
          free_time=1454368446.769452)

These can be analyzed separately or viewed in a bokeh plot using the providedvisualize method on each profiler:

>>> prof.visualize()

To view multiple profilers at the same time, the dask.diagnostics.visualizefunction can be used. This takes a list of profilers and creates a verticalstack of plots aligned along the x-axis:

>>> from dask.diagnostics import visualize
>>> visualize([prof, rprof, cprof])

Looking at the above figure, from top to bottom:

• The results from the Profiler object: This shows the execution time foreach task as a rectangle, organized along the y-axis by worker (in this casethreads). Similar tasks are grouped by color and, by hovering over each task,one can see the key and task that each block represents.
• The results from the ResourceProfiler object: This shows two lines, onefor total CPU percentage used by all the workers, and one for total memoryusage.
• The results from the CacheProfiler object: This shows a line for eachtask group, plotting the sum of the current metric in the cache againsttime. In this case it’s the default metric (count) and the lines representthe number of each object in the cache at time. Note that the grouping andcoloring is the same as for the Profiler plot, and that the taskrepresented by each line can be found by hovering over the line.From these plots we can see that the initial tasks (calls tonumpy.random.random and numpy.linalg.qr for each chunk) are runconcurrently, but only use slightly more than 100% CPU. This is because thecall to numpy.linalg.qr currently doesn’t release the Global InterpreterLock (GIL), so those calls can’t truly be done in parallel. Next, there’s a reductionstep where all the blocks are combined. This requires all the results from thefirst step to be held in memory, as shown by the increased number of results inthe cache, and increase in memory usage. Immediately after this task ends, thenumber of elements in the cache decreases, showing that they were only neededfor this step. Finally, there’s an interleaved set of calls to dot andsum. Looking at the CPU plot, it shows that these run both concurrently and inparallel, as the CPU percentage spikes up to around 350%.

Custom Callbacks

Schedulers based on dask.local.get_async (currentlydask.get, dask.threaded.get, and dask.multiprocessing.get)accept five callbacks, allowing for inspection of scheduler execution.

The callbacks are:

1. start(dsk): Run at the beginning of execution, right before thestate is initialized. Receives the Dask graph

2. start_state(dsk, state): Run at the beginning of execution, rightafter the state is initialized. Receives the Dask graph and scheduler state

3. pretask(key, dsk, state): Run every time a new task is started.Receives the key of the task to be run, the Dask graph, and the scheduler state

4. posttask(key, result, dsk, state, id): Run every time a task is finished.Receives the key of the task that just completed, the result, the Dask graph,the scheduler state, and the id of the worker that ran the task

5. finish(dsk, state, errored): Run at the end of execution, right before theresult is returned. Receives the Dask graph, the scheduler state, and a booleanindicating whether or not the exit was due to an error

Custom diagnostics can be created either by instantiating the Callbackclass with the some of the above methods as keywords or by subclassing theCallback class.Here we create a class that prints the name of every key as it’s computed:

from dask.callbacks import Callback
class PrintKeys(Callback):
    def _pretask(self, key, dask, state):
        """Print the key of every task as it's started"""
        print("Computing: {0}!".format(repr(key)))

This can now be used as a context manager during computation:

>>> from operator import add, mul
>>> dsk = {'a': (add, 1, 2), 'b': (add, 3, 'a'), 'c': (mul, 'a', 'b')}
 
>>> with PrintKeys():
...     get(dsk, 'c')
Computing 'a'!
Computing 'b'!
Computing 'c'!

Alternatively, functions may be passed in as keyword arguments to Callback:

>>> def printkeys(key, dask, state):
...    print("Computing: {0}!".format(repr(key)))
 
>>> with Callback(pretask=printkeys):
...     get(dsk, 'c')
Computing 'a'!
Computing 'b'!
Computing 'c'!

API

• dask.diagnostics.ProgressBar(minimum=0, width=40, dt=0.1, out=None)
• A progress bar for dask.

Parameters:

• minimum:int, optional

• Minimum time threshold in seconds before displaying a progress bar.Default is 0 (always display)

• width:int, optional

• Width of the bar

• dt:float, optional

• Update resolution in seconds, default is 0.1 seconds

Examples

Below we create a progress bar with a minimum threshold of 1 second beforedisplaying. For cheap computations nothing is shown:

>>> with ProgressBar(minimum=1.0):      # doctest: +SKIP
...     out = some_fast_computation.compute()

But for expensive computations a full progress bar is displayed:

>>> with ProgressBar(minimum=1.0):      # doctest: +SKIP
...     out = some_slow_computation.compute()
[########################################] | 100% Completed | 10.4 s

The duration of the last computation is available as an attribute

>>> pbar = ProgressBar()
>>> with pbar:                          # doctest: +SKIP
...     out = some_computation.compute()
[########################################] | 100% Completed | 10.4 s
>>> pbar.last_duration                  # doctest: +SKIP
10.4

You can also register a progress bar so that it displays for allcomputations:

>>> pbar = ProgressBar()                # doctest: +SKIP
>>> pbar.register()                     # doctest: +SKIP
>>> some_slow_computation.compute()     # doctest: +SKIP
[########################################] | 100% Completed | 10.4 s

• dask.diagnostics.Profiler()

• A profiler for dask execution at the task level.

  • Records the following information for each task:
  • • Key
    • Task
    • Start time in seconds since the epoch
    • Finish time in seconds since the epoch
    • Worker idExamples

>>> from operator import add, mul
>>> from dask.threaded import get
>>> dsk = {'x': 1, 'y': (add, 'x', 10), 'z': (mul, 'y', 2)}
>>> with Profiler() as prof:
...     get(dsk, 'z')
22

>>> prof.results  # doctest: +SKIP
[('y', (add, 'x', 10), 1435352238.48039, 1435352238.480655, 140285575100160),
 ('z', (mul, 'y', 2), 1435352238.480657, 1435352238.480803, 140285566707456)]

These results can be visualized in a bokeh plot using the visualizemethod. Note that this requires bokeh to be installed.

>>> prof.visualize() # doctest: +SKIP

You can activate the profiler globally

>>> prof.register()  # doctest: +SKIP

If you use the profiler globally you will need to clear out old resultsmanually.

>>> prof.clear()

• dask.diagnostics.ResourceProfiler(dt=1)

• A profiler for resource use.

  • Records the following each timestep
  • • Time in seconds since the epoch
    • Memory usage in MB
    • % CPU usageExamples

>>> from operator import add, mul
>>> from dask.threaded import get
>>> dsk = {'x': 1, 'y': (add, 'x', 10), 'z': (mul, 'y', 2)}
>>> with ResourceProfiler() as prof:  # doctest: +SKIP
...     get(dsk, 'z')
22

These results can be visualized in a bokeh plot using the visualizemethod. Note that this requires bokeh to be installed.

>>> prof.visualize() # doctest: +SKIP

You can activate the profiler globally

>>> prof.register()  # doctest: +SKIP

If you use the profiler globally you will need to clear out old resultsmanually.

>>> prof.clear()  # doctest: +SKIP

Note that when used as a context manager data will be collected throughoutthe duration of the enclosed block. In contrast, when registered globallydata will only be collected while a dask scheduler is active.

• dask.diagnostics.CacheProfiler(metric=None, metric_name=None)

• A profiler for dask execution at the scheduler cache level.

  • Records the following information for each task:
  • • Key
    • Task
    • Size metric
    • Cache entry time in seconds since the epoch
    • Cache exit time in seconds since the epochExamples

>>> from operator import add, mul
>>> from dask.threaded import get
>>> dsk = {'x': 1, 'y': (add, 'x', 10), 'z': (mul, 'y', 2)}
>>> with CacheProfiler() as prof:
...     get(dsk, 'z')
22

>>> prof.results    # doctest: +SKIP
[CacheData('y', (add, 'x', 10), 1, 1435352238.48039, 1435352238.480655),
 CacheData('z', (mul, 'y', 2), 1, 1435352238.480657, 1435352238.480803)]

The default is to count each task (metric is 1 for all tasks). Otherfunctions may used as a metric instead through the metric keyword. Forexample, the nbytes function found in cachey can be used to measurethe number of bytes in the cache.

>>> from cachey import nbytes    # doctest: +SKIP
>>> with CacheProfiler(metric=nbytes) as prof:  # doctest: +SKIP
...     get(dsk, 'z')

The profiling results can be visualized in a bokeh plot using thevisualize method. Note that this requires bokeh to be installed.

>>> prof.visualize() # doctest: +SKIP

You can activate the profiler globally

>>> prof.register()  # doctest: +SKIP

If you use the profiler globally you will need to clear out old resultsmanually.

>>> prof.clear()

• dask.diagnostics.Callback(start=None, start_state=None, pretask=None, posttask=None, finish=None)
• Base class for using the callback mechanism

Create a callback with functions of the following signatures:

>>> def start(dsk):
...     pass
>>> def start_state(dsk, state):
...     pass
>>> def pretask(key, dsk, state):
...     pass
>>> def posttask(key, result, dsk, state, worker_id):
...     pass
>>> def finish(dsk, state, failed):
...     pass

You may then construct a callback object with any number of them

>>> cb = Callback(pretask=pretask, finish=finish)  # doctest: +SKIP

And use it either as a context manager over a compute/get call

>>> with cb:  # doctest: +SKIP
...     x.compute()  # doctest: +SKIP

Or globally with the register method

>>> cb.register()  # doctest: +SKIP
>>> cb.unregister()  # doctest: +SKIP

Alternatively subclass the Callback class with your own methods.

>>> class PrintKeys(Callback):
...     def _pretask(self, key, dask, state):
...         print("Computing: {0}!".format(repr(key)))

>>> with PrintKeys():  # doctest: +SKIP
...     x.compute()  # doctest: +SKIP

• dask.diagnostics.visualize(profilers, file_path=None, show=True, save=True, **kwargs)
• Visualize the results of profiling in a bokeh plot.

If multiple profilers are passed in, the plots are stacked vertically.

Parameters:

• profilers:profiler or list

• Profiler or list of profilers.

• file_path:string, optional

• Name of the plot output file.

• show:boolean, optional

• If True (default), the plot is opened in a browser.

• save:boolean, optional

• If True (default), the plot is saved to disk.

• **kwargs

• Other keyword arguments, passed to bokeh.figure. These will overrideall defaults set by visualize.Returns:
• The completed bokeh plot object.