Short answer: - You don’t need to change the whole flow’s task runner. - The simplest way to “parallelize just a few tasks” is to use

task.submit()

to fan out and `as_completed()`/`wait()` or

.result()

to join. - If you need a different concurrency model just for that section (e.g., process-based for CPU-bound work or a different max_workers), wrap that section in a subflow that specifies its own task runner and call the subflow from your main flow. - You cannot “temporarily swap” task runners inside a single flow; task runners are configured per-flow. Why your tasks are sequential now - In Prefect 3.x, tasks run sequentially when you call them like normal functions (e.g.,

x = my_task(...)

). - To get concurrency, use

my_task.submit(...)

which returns Prefect futures. Those run under the flow’s task runner and execute concurrently. Recommended patterns 1) Fan-out/fan-in within the same flow using futures (most common)

from prefect import flow, task
from prefect.futures import as_completed

@task
def work(i):
    # do something
    return i * 2

@task
def prep():
    return "ready"

@task
def finalize(vals):
    return sum(vals)

@flow
def my_flow():
    # sequential tasks
    _ = prep()

    # run a subset concurrently
    futures = [work.submit(i) for i in range(10)]

    # consume results as they complete (or use .result() to block on all)
    results = []
    for fut in as_completed(futures):
        results.append(fut.result())

    # continue sequentially
    total = finalize(results)
    return total

2) Use a subflow with its own task runner (when you need a different concurrency model for that section) - Useful if your “parallel section” is CPU-bound (use processes) or you want a different

max_workers

.

from prefect import flow, task
from prefect.task_runners import ProcessPoolTaskRunner  # or ThreadPoolTaskRunner

@task
def cpu_heavy(i):
    # CPU-bound computation
    return i * i

@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def cpu_parallel_subflow(items):
    futs = [cpu_heavy.submit(i) for i in items]
    return [f.result() for f in futs]

@flow
def main_flow():
    # some sequential steps first...
    items = list(range(20))

    # run just this section with a different runner
    results = cpu_parallel_subflow(items)

    # continue sequentially...
    return sum(results)

Notes and tips - You cannot manually start/stop or swap a task runner mid-flow; it’s bound to the flow. - If you don’t specify a task runner, tasks still only run concurrently when you use `submit()`/`map()`; direct calls are sequential. - For large fan-outs or streaming results,

prefect.futures.as_completed()

is great for processing items as they finish. - Choose the runner based on workload: - I/O-bound: ThreadPoolTaskRunner - CPU-bound: ProcessPoolTaskRunner - Distributed: Dask or Ray runners (extras required) Docs to dig deeper - Running work concurrently: How to run work concurrently - Task runners in Prefect: Task runners concept - Futures utilities: prefect.futures API Quick question to tailor the advice: - Are the tasks you want to parallelize I/O-bound or CPU-bound? - Roughly how many parallel tasks do you need at once?

@Marvin Can you show me a code example using .result() to wait for all the futures?

Absolutely—here’s a minimal example that fans out tasks with

submit()

and then waits for all of them using `.result()`:

from prefect import flow, task

@task
def compute(i):
    # simulate some work
    return i * 2

@task
def prep():
    return "ready"

@flow
def my_flow():
    # sequential work
    _ = prep()

    # fan out concurrently
    futures = [compute.submit(i) for i in range(10)]

    # join: wait for all futures and collect results (in submission order)
    results = [f.result() for f in futures]

    # continue with sequential logic
    return sum(results)

if __name__ == "__main__":
    my_flow()

Notes: -

compute.submit(...)

returns a Prefect future and starts execution concurrently. -

f.result()

blocks until that future finishes; iterating over all futures like above waits for all of them. - If any task fails,

.result()

will raise the underlying exception. Docs for reference: - Running work concurrently: How to run work concurrently - Futures API: prefect.futures API