Task runners are not required for task execution. Calling a task function directly executes the function in the main thread by default, blocking execution of its caller until the task completes.

To enable concurrent, parallel, or distributed execution of tasks, use the .submit() or .map() methods to submit tasks to a task runner.

The default task runner in Prefect is the ThreadPoolTaskRunner, which runs tasks concurrently in independent threads.

For truly parallel or distributed task execution, you must use one of the following task runners, which are available as extras of the prefect library:

Concurrency vs. parallelism

  • Concurrency refers to a system that can do more than one thing simultaneously, but not at the exact same time. Think of concurrent execution as non-blocking: within the restrictions of resources available in the execution environment and data dependencies between tasks, execution of one task does not block execution of other tasks in a flow.
  • Parallelism refers to a system that can do more than one thing at the exact same time. Within the restrictions of resources available, parallel execution can run tasks at the same time, such as for operations mapped across a dataset.

​
Configure a task runner

To configure a specific task runner, provide a task_runner keyword argument to the parent flow:

from prefect import flow, task
from prefect.futures import wait
from prefect.task_runners import ThreadPoolTaskRunner
import time

@task
def stop_at_floor(floor):
    print(f"elevator moving to floor {floor}")
    time.sleep(floor)
    print(f"elevator stops on floor {floor}")

@flow(task_runner=ThreadPoolTaskRunner(max_workers=3))
def elevator():
    floors = []

    for floor in range(10, 0, -1):
        floors.append(stop_at_floor.submit(floor))

    wait(floors) # wait for the sequence of futures to complete

The max_workers parameter of the ThreadPoolTaskRunner controls the number of threads that the task runner will use to execute tasks concurrently.

​
Submit tasks to a task runner

When you use .submit() to submit a task to a task runner, the task runner creates a PrefectFuture for access to the state and result of the task.

A PrefectFuture is an object that provides:

  • a reference to the result returned by the task
  • a State indicating the current state of the task run

PrefectFuture objects must be resolved explicitly before returning from a flow or task. Dependencies between futures will be automatically resolved whenever their dependents are resolved. This means that only terminal futures need to be resolved, either by:

  • returning the terminal futures from your flow or task
  • calling .wait() or .result() on each terminal future
  • using one of the top level wait or as_completed utilities to resolve terminal futures

Not doing so may leave your tasks in an unfinished state.

When you pass a future into a task, Prefect automatically waits for the “upstream” task (the one that the future references), to reach a final state before starting the downstream task.

This means that the downstream task won’t receive the PrefectFuture you passed as an argument. Instead, the downstream task receives the value that the upstream task returned.

For example:

from prefect import flow, task

@task
def say_hello(name):
    return f"Hello {name}!"

@task
def print_result(result):
    print(type(result))
    print(result)

@flow(name="hello-flow")
def hello_world():
    future = say_hello.submit("Marvin")
    print_result.submit(future).wait()

hello_world()

If we run this, we see that we only had to wait for the final print_result future as Prefect automatically resolved say_hello to a string.

​
Access results from submitted tasks

You can access the result of a future explicitly with the .result() method.

from prefect import flow, task

@task
def my_task():
    return 42

@flow
def my_flow():
    future = my_task.submit()
    result = future.result()
    print(result)

my_flow()

The .result() method waits for the task to complete before returning the result to the caller. If the task run fails, .result() will raise the task run’s exception. Disable this behavior with the raise_on_failure option:

from prefect import flow, task

@task
def my_task():
    return "I'm a task!"


@flow
def my_flow():
    future = my_task.submit()
    result = future.result(raise_on_failure=False)
    if future.state.is_failed():
        # `result` is an exception! handle accordingly
        ...
    else:
        # `result` is the expected return value of our task
        ...

A few notes on .result()

  • .result() is a blocking call. This means that calling .result() will block the caller until the task run completes.
  • Only use .result() when you need to interact directly with the return value of your submitted task; for example, you should use .result() if passing the return value to a standard Python function (not a Prefect task) but do not need to use .result() if you are passing the value to another task (as these futures will be automatically resolved).

​
Creating state dependencies

You may also use the wait_for=[] parameter when calling a task by specifying upstream task dependencies. This enables you to control task execution order for tasks that do not share data dependencies.

@task
def task_a():
    pass

@task
def task_b():
    pass

@task
def task_c():
    pass
    
@task
def task_d():
    pass

@flow
def my_flow():
    a = task_a.submit()
    b = task_b.submit()
    # Wait for task_a and task_b to complete
    c = task_c.submit(wait_for=[a, b])
    # task_d will wait for task_c to complete
    # Note: If waiting for one task it must still be in a list.
    d = task_d(wait_for=[c])

​
Mapping over iterables

Prefect provides a .map() method that automatically submits a new task run for each element of its input data. This can be useful when submitting a lot of work to a task runner simultaneously.

wait_for can also be used with .map()

Similar to the .submit() method, the .map() method accepts a wait_for argument to establish state dependencies between mapped tasks that do not share data dependencies.

​
Simple mapping

from prefect import flow, task

@task
def print_nums(nums: list[int]):
    for n in nums:
        print(n)

@task
def square_num(num: int) -> int:
    return num**2

@flow
def map_flow(nums: list[int]):
    print_nums(nums)
    squared_nums = square_num.map(nums)
    print_nums(squared_nums)

map_flow([1,2,3,5,8,13])

​
Using the unmapped annotation

Sometimes you may not want to map a task over a certain input value.

By default, non-iterable values will not be mapped over (so unmapped is not required):

from prefect import flow, task

@task
def add_together(x, y):
    return x + y

@flow
def sum_it(numbers: list[int], static_value: int):
    futures = add_together.map(numbers, static_value)
    return futures.result()

resulting_sum = sum_it([1, 2, 3], 5)
assert resulting_sum == [6, 7, 8]

… but if your argument is an iterable type, wrap it with unmapped to tell .map to treat it as static:

from prefect import flow, task, unmapped

@task
def sum_plus(x, static_iterable):
    return x + sum(static_iterable)

@flow
def sum_it(numbers, static_iterable):
    futures = sum_plus.map(numbers, unmapped(static_iterable))
    return futures.result()

resulting_sum = sum_it([4, 5, 6], [1, 2, 3])
assert resulting_sum == [10, 11, 12]

​
Bulk PrefectFuture operations

When using .map as in the above example, the result of the task is a list of futures.

You can wait for or retrieve the results from these futures with wait or result methods:

futures = some_task.map(some_iterable)
results = futures.result()

which is syntactic sugar for the corresponding list comprehension:

futures = some_task.map(some_iterable)
results = [future.result() for future in futures]

​
Nested mapped tasks

To model more complex concurrent workflows, you can map tasks within other tasks:

import re

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

def count_words(text: str) -> int:
    """Count the number of words in a text."""
    return len(text.split())

def extract_emails(text: str) -> list[str]:
    return re.findall(r"[\w.+-]+@[\w-]+\.[\w.-]+", text)

@task
def analyze_texts(texts: list[str]) -> dict[str, list[int | list[str]]]:
    futures = {
        op.__name__: task(op).map(texts) for op in [count_words, extract_emails]
    }
    wait([f for futs in futures.values() for f in futs])
    return {name: [f.result() for f in futs] for name, futs in futures.items()}

@flow
def run_text_analysis():
    """Analyze a batch of social media posts with multiple operations."""
    results = analyze_texts(
        texts=[
            "Just visited #Paris! Contact me at visitor@example.com #travel #vacation",
            "Working on my new #project. Reach out at developer@example.com if interested!",
            "Happy to announce our company event #celebration #milestone email: events@company.org",
        ]
    )
    print("\nAnalysis Results:")
    print(f"  Word counts: {results['count_words']}")
    print(f"  Extracted emails: {results['extract_emails']}\n")
    return results

run_text_analysis()

This pattern is useful when you need to:

  1. Process combinations of parameters concurrently
  2. Apply multiple transformations to multiple datasets
  3. Create a grid of operations where each cell is an independent task

​
Real-world applications

Mapped tasks are particularly valuable in common data science and ETL workflows such as:

  1. Machine learning model evaluation: Train multiple models on multiple datasets concurrently
  2. ETL pipelines: Process multiple data sources with multiple transformations
  3. API data enrichment: Enrich multiple records with data from multiple external services

For example, imagine you want to find the best training configuration for a series of datasets, and you want to process all datasets concurrently:

import random
from dataclasses import dataclass

from prefect import flow, task
from prefect.futures import PrefectFuture, wait

@dataclass
class Dataset:
    name: str

@dataclass
class ModelConfig:
    name: str

@task(task_run_name="train on {dataset.name} with {model_config.name}")
def train_model(dataset: Dataset, model_config: ModelConfig) -> dict:
    return {
        "dataset": dataset.name,
        "model": model_config.name,
        "score": random.random(),
    }

@flow
def evaluate_models(datasets: list[Dataset], model_configs: list[ModelConfig]):
    all_futures: list[PrefectFuture[dict[str, object]]] = []
    for dataset in datasets:
        futures = train_model.map(
            dataset=dataset,
            model_config=model_configs,
        )
        all_futures.extend(futures)

    results = [future.result() for future in wait(all_futures).done]

    print(f"\nBest model: {max(results, key=lambda r: r['score'])}")

evaluate_models(
    datasets=[
        Dataset("customers"), Dataset("products"), Dataset("orders")
    ],
    model_configs=[
        ModelConfig("random_forest"), ModelConfig("gradient_boosting")
    ],
)

​
Use multiple task runners

Each flow can only have one task runner, but sometimes you may want a subset of your tasks to run using a different task runner than the one configured on the flow. In this case, you can create nested flows for tasks that need to use a different task runner.

For example, you can have a flow (in the example below called multiple_runner_flow) that runs its tasks locally using the ThreadPoolTaskRunner. If you have some tasks that can run more efficiently in parallel on a Dask cluster, you can create a nested flow (such as dask_nested_flow) to run those tasks using the DaskTaskRunner.

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect_dask.task_runners import DaskTaskRunner
import time


@task
def hello_local(name: str):
    time.sleep(2)
    print(f"Hello {name}!")


@flow # implicitly uses the default task runner, ThreadPoolTaskRunner
def concurrent_nested_flow():
    marvin = hello_local.submit("marvin")
    ford = hello_local.submit("ford")
    marvin.wait(), ford.wait()


@task
def hello_dask():
    print("Hello from Dask!")


@flow(task_runner=DaskTaskRunner())
def dask_nested_flow():
    hello_dask.submit().wait()


@flow # implicitly uses the default task runner, ThreadPoolTaskRunner
def parent_flow():
    concurrent_nested_flow()
    dask_nested_flow()


if __name__ == "__main__":
    parent_flow()

​
Design considerations

When choosing how and when to achieve concurrency using task runners, consider the following:

  1. Task granularity: The “proper” size for tasks depends on the nature and complexity of the work you’re doing, e.g. too many small tasks might create overhead - see Writing tasks for more.

  2. Resource constraints: Be aware of environment limitations. Using .map can create many task instances very quickly, which might exceed your resource availability.

  3. Data transfer: Large data passed between tasks can impact performance. Consider passing references to external storage when dealing with large datasets.

  4. Parallelism: For true parallelism (rather than just concurrency), consider using a specialized task runner like the DaskTaskRunner or RayTaskRunner (or propose a new task runner type).

  5. Beware of unsafe global state: Use of concurrency or parallelism features like .submit and .map must respect the underlying primitives to avoid unexpected behavior. For example, the default ThreadPoolTaskRunner runs each task in a separate thread, which means that any global state must be threadsafe. Similarly, DaskTaskRunner and RayTaskRunner are multi-process runners that require global state to be picklable.

Was this page helpful?

Securely store typed configurationLimit concurrent task runs with tags