Collecting subsets of dynamic outputs?

[erinov1]

I have a use-case where an op produces a set of DynamicOutputs indexed, e.g., by integers; the total number of outputs is determined from the op's config or a resource provided to the op. I then want to apply downstream operations to programmatically determined subsets of the DynamicOutputs.

For example, I am looking for a way to imitate the following graph using dynamic outputs:

NUM_OUTPUTS = 5

@op()
def one():
    return 1

@op()
def add(x):
    return sum(x)

@graph():
    outputs = [one() for i in range(1, NUM_OUTPUTS)]
    for i in range(1, NUM_OUTPUTS):
        add(outputs[:i])

In the last loop I am applying the add op to the first output, then to the first and second outputs, etc. The problem in my actual use-case is that the total number of outputs (equal to NUM_OUTPUTS = 5 in the example) is not known during graph construction (or rather it is encoded in a config that I wouldn't want to load externally from dagit just for the purpose of graph construction).

Is there any creative way to achieve this?

(My actual problem involves doing expanding-window validation for an ML time-series problem on a training set of configurable length, but I think the example encompasses the main idea.)

[alangenfeld]

imitate the following graph using dynamic outputs:

You could have an op lets call it windows that dynamically returns the window values based on its input config:

@op(out=DynamicOut(), config_schema={'num_outputs': int})
def windows(context):
    num_outs = context.op_config['num_outputs']
    outputs = [1 for i in range(1, num_outs)]
    for i in range(1, num_outs):
        yield DynamicOutput(outputs[:i], mapping_key=f"window_to_{i}")

@op
def add(x):
    return sum(x)

@graph
def example():
    windows().map(add)

for your real use case, you may want the windows op to just return Tuple[int, int] and do the loading / window creation in the downstream ops.

[erinov1]

Thanks @alangenfeld. I realize my example formulation did not necessarily illustrate the functionality I'm trying to imitate, so maybe it's easier explain in words.

I have root job that produces a list of outputs, e.g., a list of partitions of some set of data. For each partition, I would like to perform an operation, e.g., training a model. This is feasible using existing map functionality. However, in a further downstream step I would like to collect subsets of the models and process them further in a nontrivial way.

For example, suppose the root job yields three partitions. The next step would produce three models, [model_1, model_2, model_3], one for each partition. In the last step I would want to call a function, call it process_models, that acts on the following subsets:

[model_1] ----> process_models()

[model_1, model_2] ----> process_models()

[model_1, model_2, model_3] ----> process_models()

One option is to collect the models, have an op that dynamically outputs the sliding windows (as you suggest) and then map using process_models. But this is not parallelized in the sense that I would have to wait for all three models [model_1, model_2, model_3] to finish training just to call

[model_1] ----> process_models()

etc. Is this possible?

On a related note, while I have seen this discussed before, I'm guessing there are no immediate plans to allow the graph topology to depend on configs? (I have other use cases apart from this example.)

[alangenfeld]

I don't believe it's currently possible to achieve that within one job run.

no immediate plans to allow the graph topology to depend on configs

Correct.

A couple of potential paths:

• Dynamically create the different jobs/graphs at definition time https://docs.dagster.io/concepts/ops-jobs-graphs/jobs-graphs#patterns
• Chain together smaller jobs using something like an asset sensor https://docs.dagster.io/concepts/partitions-schedules-sensors/sensors#multi-asset-sensors

[erinov1]

• Dynamically create the different jobs/graphs at definition time https://docs.dagster.io/concepts/ops-jobs-graphs/jobs-graphs#patterns

In principle I am happy to split up the work across different jobs. I took a look at the linked section, but it seems to me like I would still need to know a priori the number of initial ops that are being dynamically spawned. Did you have a specific thought about how explicit GraphDefinitions could be used here?

[alangenfeld]

No specific thoughts - just liking to some ideas that I thought had a chance at being useful