Great question! Assuming you’re on Prefect 3.x with the Dask integration (

prefect-dask

), you set “resources” at the Dask worker level and require them from tasks using

dask.annotate

. Quick start: - Use a LocalCluster via

DaskTaskRunner(cluster_class="distributed.LocalCluster", ...)

- Give workers resource tokens (e.g., GPU slots) with

resources

(often via

worker_kwargs

) - Wrap task submissions in

dask.annotate(resources={...})

so Dask schedules them appropriately Example for a local PC:

# pip install "prefect[dask]"

import dask
from prefect import flow, task
from prefect_dask import DaskTaskRunner

@task
def gpu_task(x):
    return x * 2

@task
def cpu_task(x):
    return x + 1

@flow(
    task_runner=DaskTaskRunner(
        # Use a local Dask cluster
        cluster_class="distributed.LocalCluster",
        cluster_kwargs={
            "n_workers": 2,
            "threads_per_worker": 2,
            "memory_limit": "4GB",         # per worker
            # Safest way to assign resources to each worker
            "worker_kwargs": {"resources": {"GPU": 1, "process": 1}},
            # Optional: disable dashboard if you don't want it locally
            # "dashboard_address": None
        },
    )
)
def my_flow():
    results = []
    # Require a GPU token for these tasks
    with dask.annotate(resources={"GPU": 1}):
        for i in range(4):
            results.append(gpu_task.submit(i))

    # Constrain these tasks using a custom resource
    with dask.annotate(resources={"process": 1}):
        for i in range(4):
            results.append(cpu_task.submit(i))

    return [r.result() for r in results]

if __name__ == "__main__":
    my_flow()

Notes: - Resources are abstract tokens; Dask doesn’t auto-detect GPUs/limits. You define what each worker “has” (e.g.,

{"GPU": 1}

), and tasks declare what they “need” via

dask.annotate(resources={...})

. - You can also set general concurrency with

n_workers

,

threads_per_worker

, and memory via

memory_limit

. - If you prefer, some Dask versions accept

resources

directly in

LocalCluster(...)

. Using

worker_kwargs={"resources": {...}}

is a safe, universal pattern for LocalCluster. Helpful docs: - Prefect Dask integration: Prefect Dask docs - Dask resource scheduling: Dask resources guide If you share what resource you want to enforce (e.g., GPU slots, RAM gating, or a custom token), I can tailor the snippet exactly to that.