Parallel Chaining of Dynamic Ops in Dagster

[diogo-dil]

What's the issue?

I'm using Dagster v1.10 and trying to build a job that performs the following steps:

1. Dynamically generates a list of time-based partitions (e.g., per minute).
2. For each partition:
   • Processes the partition
   • Immediately uploads the result

Desired Execution Flow

generate_partitions() ──> process_partition() ──> upload_partition()
                          (runs in parallel)

Current setup

@op(out=DynamicOut())
def generate_partitions(context):
    partitions = ["2025-07-01T12:00", "2025-07-01T12:01", "2025-07-01T12:02"]
    for p in partitions:
        yield DynamicOutput(value=p, mapping_key=p.replace(":", "_"))

@op
def process_partition(context, partition: str) -> str:
    return f"{partition}_processed"

@op
def upload_partition(context, processed_data: str):
    context.log.info(f"Uploading {processed_data}")


@job
def partition_processing_job():
    dynamic_partitions = generate_partitions()
    processed = dynamic_partitions.map(process_partition)
    uploaded = processed.map(upload_partition)

note:

I tried to use the graph but no success as well...

Problem

Dagster waits for all dynamic outputs from generate_partitions and all process_partition ops to complete before running any upload_partition op.
But I want each upload_partition to start as soon as its corresponding process_partition finishes.

What did you expect to happen?

How can I get Dagster to execute the upload step as soon as its matching processing step finishes, instead of waiting for all partitions to complete first?

Is this possible in Dagster v1.10?

[martincpt]

I would love to this as well.

[zyd14]

I haven't done it myself but I think I've seen people on the Dagster slack solve this by putting the steps they want to complete in sequence in a nested graph. So in your case process_partition and upload_partition would go in a @graph, and you'd map dynamic_partitions over that graph. Something like this:

@op(out=DynamicOut())
def generate_partitions(context):
    partitions = ["2025-07-01T12:00", "2025-07-01T12:01", "2025-07-01T12:02"]
    for p in partitions:
        yield DynamicOutput(value=p, mapping_key=p.replace(":", "_"))

@op
def process_partition(context, partition: str) -> str:
    return f"{partition}_processed"

@op
def upload_partition(context, processed_data: str):
    context.log.info(f"Uploading {processed_data}")

@graph
def partition_processing_graph(partition_input):
  return upload_partition(process_partition(partition_input)) 

@job
def partition_processing_job():
    dynamic_partitions = generate_partitions()
    processed = dynamic_partitions.map(partition_processing_graph)

I'm not sure if that syntax is correct, it might be necessary to use .map() within the nested graph as well.