Great question. In Prefect 3.x with RayTaskRunner, there are two good ways to cap concurrency of a mapped task to 10: Option 1 — Prefect-native concurrency limit (recommended; works regardless of Ray cluster size) - Tag the task, then create a concurrency limit for that tag. Prefect will ensure no more than 10 runs of that task are in the Running state at once. Code:

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner

@task(tags=["limited-10"])
def my_task(x):
    # do work
    return x

@flow(task_runner=RayTaskRunner())
def my_flow():
    my_task.map(range(100))

if __name__ == "__main__":
    my_flow()

CLI (verified):

prefect concurrency-limit create limited-10 10

You can inspect or adjust later with:

prefect concurrency-limit inspect limited-10
prefect concurrency-limit ls
prefect concurrency-limit reset limited-10
prefect concurrency-limit delete limited-10

Option 2 — Limit Ray parallelism by resources - If you’re running Ray locally, you can cap available CPUs so only 10 1-CPU tasks run at once:

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner

@task
def my_task(x):
    return x

@flow(task_runner=RayTaskRunner(init_kwargs={"num_cpus": 10}))
def my_flow():
    my_task.map(range(100))

if __name__ == "__main__":
    my_flow()

- If you’re on a larger/remote Ray cluster and still want only 10, either: - Use Prefect’s tag-based concurrency limit (Option 1), or - Use a custom Ray resource with capacity 10 and require it per task via

prefect_ray.context.remote_options(resources={"limited": 1})

. Quick note: there’s no built-in

max_concurrency

parameter on

.map()

itself; use one of the approaches above. If you can share whether you’re using a local Ray runtime or a remote cluster, I can tailor the example further (e.g., using

remote_options

to control per-task resource requirements).

Short answer: - Don’t use

with_options

to control Ray parallelism; that configures Prefect options, not Ray scheduling. - Use the Ray context manager from Prefect:

prefect_ray.context.remote_options(...)

. Tasks submitted inside the block (including

.map

) inherit those Ray options. Two dynamic patterns you can use for Option 2: A) Cap concurrency by per-task CPU share (works on any Ray cluster) - Compute how many CPUs the cluster has at runtime, then choose

num_cpus

so at most

limit

tasks can run in parallel.

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options
import ray

@task
def my_task(x): 
    return x

@flow(task_runner=RayTaskRunner())
def my_flow(limit: int = 10):
    # Ray is initialized by the runner; now we can query the cluster
    total_cpus = float(ray.cluster_resources().get("CPU", 1.0))
    per_task_cpus = max(0.001, total_cpus / float(limit))  # avoid zero
    with remote_options(num_cpus=per_task_cpus):
        my_task.map(range(100))

if __name__ == "__main__":
    my_flow(limit=10)

This keeps sum(num_cpus of running tasks) ≤ total CPUs, so you get ~`limit` tasks in flight. B) Use a custom Ray resource token pool (precise cap at N) - Start Ray with a custom resource of capacity

limit

, then require 1 unit per task. - You can set this dynamically per run by injecting a task runner at call time with

Flow.with_options

.

from prefect import flow, task
from prefect_ray.task_runners import RayTaskRunner
from prefect_ray.context import remote_options

@task
def my_task(x): 
    return x

@flow
def my_flow(limit: int, n: int = 100):
    # Require one 'limited' token per task
    with remote_options(resources={"limited": 1}):
        my_task.map(range(n))

def run(limit: int):
    # Dynamically size the local Ray runtime for this run
    runner = RayTaskRunner(init_kwargs={"resources": {"limited": float(limit)}})
    # Inject the runner at call time
    my_flow.with_options(task_runner=runner)(limit=limit)

if __name__ == "__main__":
    run(limit=10)

Notes -

with_options

on Flow/Task does not set Ray options; it can be used to inject a different

task_runner

at call time (as shown above). - If you’re on a remote/shared Ray cluster, you generally can’t change its resources per run. Prefer pattern A (per-task

num_cpus

) in that case. - For local Ray via

RayTaskRunner

, you can also set a hard cap by CPUs:

RayTaskRunner(init_kwargs={"num_cpus": limit})

and inject it with

Flow.with_options(...)

similarly. If you share whether you’re on a local Ray runtime or a remote cluster, I can tailor the snippet to your setup.