The ability for portions of a graph to be duplicated at runtime.
Name | Description |
---|---|
DynamicOut | Declare that an op will return dynamic outputs |
DynamicOutput | The object that an op will yield repeatedly, each containing a value and a unique mapping_key |
The basic unit of computation in Dagster is the op. In certain cases it is desirable to run the same op multiple times on different pieces of similar data.
Dynamic outputs are the tool Dagster provides to allow resolving the pieces of data at runtime and having downstream copies of the ops created for each piece.
Here we start with a contrived example of a job containing a single expensive op:
@op
def data_processing():
large_data = load_big_data()
interesting_result = expensive_processing(large_data)
return analyze(interesting_result)
@job
def naive():
data_processing()
While, the implementation of expensive_computation
can internally do something to parallelize the compute, if anything goes wrong with any part we have to restart the whole computation.
With this motivation we will break up the computation using Dynamic Outputs. First we will define our new op that will use dynamic outputs. First we use DynamicOut
to declare that this op will return dynamic outputs. Then in the function we yield
a number of DynamicOutput
objects that each contain a value and a unique mapping_key
.
@op(out=DynamicOut())
def load_pieces():
large_data = load_big_data()
for idx, piece in large_data.chunk():
yield DynamicOutput(piece, mapping_key=idx)
Then after creating ops for our downstream operations, we can put them all together in a job.
@op(out=DynamicOut())
def load_pieces():
large_data = load_big_data()
for idx, piece in large_data.chunk():
yield DynamicOutput(piece, mapping_key=idx)
Within our @job
decorated composition function, the object representing the dynamic output can not be passed directly to another op. Either map
or collect
must be invoked on it.
map
takes a Callable
which receives a single argument. This callable is evaluated once, and any invoked op that is passed the input argument will establish dependencies. The ops downstream of a dynamic output will be cloned for each dynamic output, and identified using the associated mapping_key
. The return value from the callable is captured and wrapped in an object that allows for subsequent map
or collect
calls.
collect
creates a fan-in dependency over all the dynamic copies. The dependent op will receive a list containing all the values.
The following two examples are equivalent ways to establish a sequence of ops that occur for each dynamic output.
@job
def chained():
results = dynamic_values().map(echo).map(echo).map(echo)
process(results.collect())
@job
def chained_alt():
def _for_each(val):
a = echo(val)
b = echo(a)
return echo(b)
results = dynamic_values().map(_for_each)
process(results.collect())
A lambda or scoped function can be used to pass non-dynamic outputs along side dynamic ones in map
downstream.
@job
def other_arg():
non_dynamic = one()
dynamic_values().map(lambda val: add(val, non_dynamic))
Multiple outputs are returned via a namedtuple
, where each entry can be used via map
or collect
.
@job
def multiple():
# can unpack on assignment (order based)
values, _ = multiple_dynamic_values()
process(values.collect())
# or access by name
outs = multiple_dynamic_values()
process(outs.values.collect())