Great question! In Prefect 3.x,

.map

fans out many task runs concurrently. You control how many of those run at once with one or more of the following: 1) Limit concurrency within a single flow run (task runner) - Set the task runner on your flow and cap workers. This directly bounds how many mapped items run in parallel in that flow.

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

@task
def do_work(x):
    ...

@flow(task_runner=ThreadPoolTaskRunner(max_workers=5))  # at most 5 mapped tasks at once
def my_flow(items):
    futures = do_work.map(items)
    return [f.result() for f in futures]

- Use threads for I/O-bound work; processes for CPU-bound. - Docs: Task runners 2) Limit concurrency across flows with tag-based limits - Add a tag to the task; set a concurrency limit for that tag. This caps how many tasks with that tag can run at once across your entire Prefect environment.

from prefect import flow, task

@task(tags=["db"])
def query_db(q):
    ...

@flow
def my_flow(queries):
    futures = query_db.map(queries)
    return [f.result() for f in futures]

Create the limit:

prefect concurrency-limit create db 3

- Concepts: Tag-based concurrency limits - How-to: How to limit concurrent task runs with tags 3) Apply global concurrency/rate limits inside the task - Use a global concurrency lease so your task blocks until a slot is available (works for sync or async and can be used for true rate limiting). Sync example:

from prefect import task
from prefect.concurrency.sync import concurrency

@task
def call_api(x):
    with concurrency("api-calls", occupy=1):
        return do_the_call(x)

Create the limit:

prefect global-concurrency-limit create api-calls --limit 5

- Concepts: Global concurrency limits - How-to: How to apply global concurrency and rate limits 4) Distributed runners (optional) - If you’re using Dask, concurrency is governed by the cluster’s worker/threads configuration via

DaskTaskRunner

. - Docs: Task runners 5) Flow-level concurrency (work pools) - Limits how many flow runs (not individual task runs) execute at once. Helpful if many flows each

.map

lots of work.

prefect work-pool update my-pool --concurrency-limit 5

- Concepts: Work pools - How-to: Manage work pools Also see: - How to run work concurrently (covers

.map

and

.submit

patterns) Quick guidance: - Want to cap concurrency only within this flow run? Use the flow’s task runner

max_workers

. - Need to coordinate access to a shared resource across many flows? Use tag-based or global concurrency limits (or both). - Need distributed throughput? Use

DaskTaskRunner

and size the cluster accordingly. Would you like to cap concurrency just within a single flow run, or enforce a limit across multiple flows/environments? That’ll help me recommend the best option and an exact snippet for your case.

In Prefect 3.x, the default task runner is ThreadPoolTaskRunner. By default it uses a very high max_workers (effectively unlimited, based on sys.maxsize) unless you set it explicitly on the flow. Docs: Task runners