Custom Graphs
Contents
Custom Graphs¶
There may be times when you want to do parallel computing but your application doesn’t fit neatly into something like Dask Array or Dask Bag. In these cases, you can interact directly with the Dask schedulers. These schedulers operate well as standalone modules.
This separation provides a release valve for complex situations and allows advanced projects to have additional opportunities for parallel execution, even if those projects have an internal representation for their computations. As Dask schedulers improve or expand to distributed memory, code written to use Dask schedulers will advance as well.
Example¶
As discussed in the motivation and specification sections, the schedulers take a task graph (which is a dict of tuples of functions) and a list of desired keys from that graph.
Here is a mocked out example building a graph for a traditional clean and analyze pipeline:
def load(filename):
...
def clean(data):
...
def analyze(sequence_of_data):
...
def store(result):
with open(..., 'w') as f:
f.write(result)
dsk = {'load-1': (load, 'myfile.a.data'),
'load-2': (load, 'myfile.b.data'),
'load-3': (load, 'myfile.c.data'),
'clean-1': (clean, 'load-1'),
'clean-2': (clean, 'load-2'),
'clean-3': (clean, 'load-3'),
'analyze': (analyze, ['clean-%d' % i for i in [1, 2, 3]]),
'store': (store, 'analyze')}
from dask.threaded import get
get(dsk, 'store') # executes in parallel
Keyword arguments in custom Dask graphs¶
Sometimes, you may want to pass keyword arguments to a function
in a custom Dask graph.
You can do that using the dask.utils.apply()
function, like this:
from dask.utils import apply
task = (apply, func, args, kwargs) # equivalent to func(*args, **kwargs)
dsk = {'task-name': task,
...
}
In the example above:
args
should be a tuple (eg:(arg_1, arg_2, arg_3)
), andkwargs
should be a dictionary (eg:{"kwarg_1": value, "kwarg_2": value}