Thanks for the detailed context — I see the Dask bug you linked (https://github.com/PrefectHQ/prefect/issues/16904) and the pressure it puts on you to mark the shared helper with

@task

. The big jump in task count is a classic “task fan‑out” effect: once a frequently‑called helper is a task, every call becomes a scheduled task unless you explicitly call it as a plain function. Here are practical ways to bring your task count (and runtime) back down while keeping the Dask workaround: 1) Call the underlying function with .fn when you don’t need a separate task - A Prefect task can be invoked three ways: -

helper(...)

inside a flow: creates a task run -

helper.submit(...)

inside a flow: creates a task run and schedules it for parallel execution -

helper.fn(...)

anywhere: calls the underlying Python function directly (no task run) - Change high‑frequency calls to

helper.fn(...)

so they don’t create thousands of tasks. Keep only the calls you truly want orchestrated/parallelized as

helper.submit(...)

. Example pattern

from prefect import flow
from prefect.tasks import task
from prefect_dask import DaskTaskRunner

@task
def helper(x):
    # shared helper logic
    return x + 1

@task
def process_batch(batch):
    # Use .fn to avoid per-item tasks
    return [helper.fn(i) for i in batch]

@flow(task_runner=DaskTaskRunner())
def main(items):
    # Chunk to control parallelism
    batches = [items[i:i+200] for i in range(0, len(items), 200)]
    # Only N tasks here, one per batch
    results = [process_batch.submit(b) for b in batches]
    return [r.result() for r in results]

Key tradeoff:

helper.fn(...)

skips retries, caching, and result persistence for that call. If you need those features, keep the call as a task (`helper(...)`/`.submit(...)`) — but use batching to limit how many tasks you create. 2) Batch work to reduce task explosion - Wrap loops in a coarser “batch” task instead of calling the helper task once per item. - A good heuristic is to target a few hundred to a thousand total tasks for a run (depending on result size and environment), not thousands per small helper call. 3) Turn off result persistence and heavy logging for high‑cardinality tasks - If you do keep many tasks, disable result persistence and extra logs on the “noisy” ones:

@task(persist_result=False, cache_result_in_memory=False, log_prints=False)
def helper(x):
    ...

- This reduces storage, serialization, and UI overhead. 4) Make sure your shared module is present on all Dask workers so you don’t need @task If the only reason you added

@task

was to ship the function to Dask workers, the most robust fix is to package/install the shared module in the worker environment. Then you can remove

@task

from the helper entirely and treat it as a regular import: - Build a runtime image (Docker/conda) that installs your shared module (e.g.,

pip install .

or

pip install -e .

) and use that for the workers. - Or otherwise ensure every Dask worker has the same environment and can import your shared code. - This avoids the Dask import issue and eliminates the need to decorate helpers as tasks in the first place. Docs and references - Dask integration and runner setup: Prefect Dask integration docs - Prefect 3 tasks API (source for

task

options like persist_result): prefect/tasks.py source Questions to help tailor the fix - Where is the helper being called — inside tight loops, list comprehensions, or comprehensions inside other tasks? - Do you need retries/caching at the helper level, or only at a higher “batch”/“stage” level? - Are you using

.submit

widely (which will also create many tasks)? - How is your Dask cluster configured — ephemeral per flow run or a long‑lived cluster you connect to? Do all workers share the same environment?

If you can share a minimal snippet showing the helper definition and where it’s called, I can suggest an exact refactor to minimize task count and keep the Dask workaround in place.

That warning means Prefect tried to acquire a tag-based concurrency slot for one or more tags on your run, but no matching global concurrency limit exists. When that happens, acquisition is skipped and the run proceeds without limiting. In Prefect 3.x, tag-based concurrency works by creating a Global Concurrency Limit whose name is prefixed with

tag:

and matching it to a tag you apply to your flows/tasks. For example, for the tag

image_ingest1

, you must create a GCL named

tag:image_ingest1

. How to fix it 1) Create the limit (CLI) - Create a limit of 3 for your tag:

prefect gcl create tag:image_ingest1 --limit 3

- List limits:

prefect gcl ls

- Inspect this limit:

prefect gcl inspect tag:image_ingest1

- Delete it (if needed):

prefect gcl delete tag:image_ingest1

2) Tag your flow/tasks so they map to the limit Add the tag

image_ingest1

in your code. Prefect will look for a GCL named

tag:image_ingest1

.

from prefect import flow, task, tags

@task
def ingest_one(item):
    # your work here
    ...

@flow
def image_pipeline(items):
    # All work inside this context is tagged "image_ingest1"
    with tags("image_ingest1"):
        for i in items:
            ingest_one.submit(i)

Common causes of this warning - The GCL wasn’t created yet. - Name mismatch: the limit must be named exactly

tag:<your_tag>

, e.g.,

tag:image_ingest1

. - You created the limit in a different workspace or against a different API URL than the one your runs are using. Docs - Tag-based concurrency limits: Tag-based concurrency limits - Managing global concurrency limits (CLI/API): Global concurrency limits If you share how you’re tagging your flow/task and what you named your limit, I can help verify the mapping.