Yes—this is a great fit for Prefect’s task dependency model. In Prefect 3.x you can submit tasks in two parallel branches, then “join” them with a final task that runs only after both branches complete. Key points: - Tasks submitted with

.submit()

run concurrently under the flow’s task runner. - Passing

PrefectFuture

objects as arguments to another task automatically creates dependencies. - If your final task doesn’t need the results as inputs, you can make dependencies explicit with the

wait_for

kwarg. Example:

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner  # alias of ThreadPoolTaskRunner

@task
def branch1_task(x):
    return f"b1:{x}"

@task
def branch2_task(y):
    return f"b2:{y}"

@task
def finalize(results1, results2):
    print("Final got:", results1, results2)
    return "done"

@task
def finalize_no_inputs():
    return "done"

@flow(task_runner=ConcurrentTaskRunner(max_workers=8))
def pipeline():
    # Branch 1 — runs in parallel
    b1 = [branch1_task.submit(i) for i in range(3)]

    # Branch 2 — runs in parallel
    b2 = [branch2_task.submit(i) for i in range(3)]

    # Join by passing futures as args (dependencies inferred automatically)
    joined = finalize.submit(b1, b2)

    # Or, if the final task takes no inputs, make dependencies explicit:
    # joined = finalize_no_inputs.submit(wait_for=[*b1, *b2])

    return joined.result()

if __name__ == "__main__":
    pipeline()

Notes: - The default task runner is a thread pool; using

ConcurrentTaskRunner(max_workers=...)

lets you control concurrency. - Passing futures as arguments will resolve them to their results inside

finalize

, and ensures

finalize

won’t run until both branches finish. If you’re not using Prefect (or if you’re on a different version), let me know your setup and I can tailor the example.

Got it — for Prefect 2.8.0 this is fully supported. How to do it: - Submit tasks with

.submit()

to run them in parallel (default

ConcurrentTaskRunner

in 2.8.0). - “Join” the branches with a final task by either: - Passing the futures as arguments (dependencies inferred automatically), or - Using the

wait_for

kwarg if the final task doesn’t need the results as inputs. Example (both patterns shown):

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def b1(x): return f"b1:{x}"

@task
def b2(y): return f"b2:{y}"

@task
def finalize(group1, group2):
    # group1/group2 will be lists of resolved results
    return {"g1": group1, "g2": group2}

@task
def finalize_no_inputs():
    return "done"

@flow(task_runner=ConcurrentTaskRunner())
def pipeline():
    # Branch 1 (parallel)
    g1 = [b1.submit(i) for i in range(3)]
    # Branch 2 (parallel)
    g2 = [b2.submit(i) for i in range(3)]

    # Option A: pass futures as args (dependencies inferred automatically)
    joined = finalize.submit(g1, g2)

    # Option B: if final task has no inputs, make dependencies explicit
    # joined = finalize_no_inputs.submit(wait_for=[*g1, *g2])

    # Only call .result() when you need to block for the value
    return joined.result()

if __name__ == "__main__":
    pipeline()

Notes: - Passing `PrefectFuture`s as args automatically creates upstream dependencies and resolves to values before the final task runs. - If you see tasks running sequentially, double-check that you’re not calling

.result()

too early and that you’re not using

SequentialTaskRunner

. If you’re planning ongoing development, consider upgrading to Prefect 3.x for the latest features — but the above works great in 2.8.0.

Yes. In Prefect 2.8.0, using the default

ConcurrentTaskRunner

, every call to

.submit()

schedules the task to run concurrently. In the example: - All

b1.submit(...)

calls are concurrent with each other. - All

b2.submit(...)

calls are concurrent with each other. - The

b1

and

b2

groups can run at the same time (there’s no barrier between them). - The final task won’t run until both groups finish (either because you pass the futures as args or use

wait_for

). A few gotchas: - Don’t call

.result()

on the branch futures before submitting the other branch or the final task; that would serialize execution. - If you switch to

SequentialTaskRunner

, tasks run one-at-a-time. - For heavy CPU-bound work, consider a process-based/distributed runner (e.g., Dask) for true parallelism.

Yes—totally doable in Prefect 2.8.0. You make each group a chain (each task depends on the previous), and submit both chains so they run concurrently under the default ConcurrentTaskRunner. Two ways to chain within a group: 1) Pass the previous task’s future as an argument (dependencies inferred)

from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def step(name, prev=None):
    return f"{prev}->{name}" if prev else name

@task
def finalize():
    return "done"

@flow(task_runner=ConcurrentTaskRunner())
def run():
    # Chain A (sequential)
    a1 = step.submit("A1")
    a2 = step.submit("A2", prev=a1)
    a3 = step.submit("A3", prev=a2)
    a4 = step.submit("A4", prev=a3)
    a5 = step.submit("A5", prev=a4)
    a6 = step.submit("A6", prev=a5)

    # Chain B (sequential)
    b1 = step.submit("B1")
    b2 = step.submit("B2", prev=b1)
    b3 = step.submit("B3", prev=b2)
    b4 = step.submit("B4", prev=b3)
    b5 = step.submit("B5", prev=b4)
    b6 = step.submit("B6", prev=b5)

    # Optional: join at the end (don’t block earlier to preserve concurrency)
    done = finalize.submit(wait_for=[a6, b6])
    return done.result()

2) Or use

wait_for

explicitly (useful if tasks don’t take the previous result as an input)

@flow(task_runner=ConcurrentTaskRunner())
def run():
    # Chain A
    a1 = step.submit("A1")
    a2 = step.submit("A2", wait_for=[a1])
    a3 = step.submit("A3", wait_for=[a2])
    a4 = step.submit("A4", wait_for=[a3])
    a5 = step.submit("A5", wait_for=[a4])
    a6 = step.submit("A6", wait_for=[a5])

    # Chain B
    b1 = step.submit("B1")
    b2 = step.submit("B2", wait_for=[b1])
    b3 = step.submit("B3", wait_for=[b2])
    b4 = step.submit("B4", wait_for=[b3])
    b5 = step.submit("B5", wait_for=[b4])
    b6 = step.submit("B6", wait_for=[b5])

    finalize.submit(wait_for=[a6, b6]).result()

Tips: - Don’t call

.result()

on intermediate steps; that would serialize everything. - This gives “A1->A2->...->A6” and “B1->B2->...->B6” each sequentially, while both chains proceed concurrently with each other. - For heavy CPU-bound work, consider a process-based/distributed runner (e.g., Dask via prefect-dask).