Great question! In Prefect 3.x there are two common ways to cap how many tasks run at the same time: 1) Limit concurrency within a single flow run (simple, local) - Use a task runner with a max_workers setting. This caps how many tasks can execute concurrently in that flow run. Example for I/O-bound work (threads):

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

@task
def fetch(i):
    # do network or other I/O
    return i

@flow(task_runner=ThreadPoolTaskRunner(max_workers=3))
def my_flow():
    futures = [fetch.submit(i) for i in range(10)]
    return [f.result() for f in futures]

my_flow()

Example for CPU-bound work (processes):

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

@task
def crunch(i):
    # CPU-heavy work
    return sum(j*j for j in range(i))

@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def my_flow():
    futures = [crunch.submit(i * 1_000_000) for i in range(10)]
    return [f.result() for f in futures]

my_flow()

Docs: - Task runners concept - How to run work concurrently 2) Limit concurrency across flows/workers (server-coordinated) - If you need limits that apply across multiple flows or machines (e.g., shared DB/API), use Prefect’s server-side limits: - Tag-based concurrency limits (tag tasks and set a max running at once) - Global concurrency limits (fine-grained, can throttle any code region, supports rate limiting) Docs: - Tag-based concurrency limits (how-to) - Global concurrency and rate limits (how-to) Quick check: Do you want to limit tasks just within a single flow run, or enforce a limit across multiple flows/workers?

Short answer: Yes—limits still apply with DaskTaskRunner, but how you set them differs. - Limiting within a single flow run (local): With DaskTaskRunner there’s no

max_workers

on the runner. You cap parallelism by configuring the Dask cluster (workers and threads). For example:

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

@task
def work(i): 
    return i

@flow(task_runner=DaskTaskRunner(
    cluster_kwargs={
        "n_workers": 2,           # processes
        "threads_per_worker": 2   # threads per process
    }
))
def my_flow():
    futs = [work.submit(i) for i in range(10)]
    return [f.result() for f in futs]

You can also use adaptive scaling:

@flow(task_runner=DaskTaskRunner(adapt_kwargs={"minimum": 1, "maximum": 4}))
def my_flow(): ...

- Limiting across flows/workers (server-coordinated): Tag-based concurrency limits and Global Concurrency Limits work regardless of task runner (including Dask). Prefect will hold task runs until slots are available. Tag-based example:

from prefect import flow, task

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

@flow
def pipeline():
    futs = [query.submit("SELECT 1") for _ in range(10)]
    return [f.result() for f in futs]

# Set limit once:
# prefect concurrency-limit create db 3

Global concurrency example (works with any runner):

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

@task
def call_api():
    with concurrency("api", occupy=1):
        # do the request
        ...

Docs: - Prefect Dask integration - Task runners concepts - Tag-based concurrency limits - Global concurrency and rate limits If you share how you’re running Dask (LocalCluster vs. a remote scheduler) and what limit you need (within a flow vs. cross-flows), I can suggest the exact config.