Great question. In Prefect 3.x, task caching relies on persisted task results. To make caches persist across Docker containers, you need to: 1) Persist results (per task or globally), and 2) Mount a host or named Docker volume at the directory where Prefect stores results inside the container. Key settings: -

PREFECT_RESULTS_PERSIST_BY_DEFAULT

(default: false) — turn on persistence globally -

PREFECT_LOCAL_STORAGE_PATH

(default: ~/.prefect/storage) — where local results are written - If you don’t set a default result storage block, Prefect uses LocalFileSystem at

PREFECT_LOCAL_STORAGE_PATH

What to do for a Docker worker - Mount a volume to the container path you’ll use for

PREFECT_LOCAL_STORAGE_PATH

(or just mount to the default

~/.prefect/storage

) - Ensure results are persisted (

persist_result=True

on tasks or set

PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

) Example: per-deployment (prefect.yaml)

deployments:
  - name: cached
    entrypoint: flows/my_flow.py:my_flow
    work_pool:
      name: docker-pool
      job_variables:
        image: prefecthq/prefect:3-latest
        volumes:
          # Host path or named volume -> container path
          - /var/prefect/cache:/home/prefect/.prefect/storage
          # or: my-prefect-cache:/home/prefect/.prefect/storage
        env:
          PREFECT_RESULTS_PERSIST_BY_DEFAULT: "true"
          PREFECT_LOCAL_STORAGE_PATH: /home/prefect/.prefect/storage

Example: task with caching

from prefect import task, flow
from prefect.cache_policies import INPUTS
from datetime import timedelta

@task(persist_result=True, cache_policy=INPUTS, cache_expiration=timedelta(days=1))
def slow_add(x, y):
    return x + y

@flow
def my_flow():
    a = slow_add(1, 2)
    b = slow_add(1, 2)  # served from cache if persisted
    return a, b

Create pool and deploy

prefect work-pool create docker-pool --type docker
prefect deploy --name cached --entrypoint flows/my_flow.py:my_flow

Notes - There is no separate “cache key directory” to configure; cache keys/metadata live in the Prefect API/database. What you persist on disk are the task results that caching can reuse. - If you prefer cross-host persistence without Docker volumes, use a remote result storage block (e.g., S3/GCS) and set that as your default result storage. - Make sure the container user has write permissions to the mounted path. Docs and references - Docker deployments and job variables (env/volumes): Docker deployment guide - Customize job variables: Customize job variables - Settings and profiles: Settings and profiles - Manage settings: Manage settings - Results (persistence): Results guide - Caching: Caching concept - Work pools: Work pools

Thanks for sharing your job template — the key piece you’re missing for private GCR/Artifact Registry images is proper Docker registry authentication. A few important points and a quick fix: Key points - The env var

GOOGLE_APPLICATION_CREDENTIALS

inside your flow container does not help with pulling the image. The worker must authenticate to the Docker registry before the pull. - In Prefect 3 with the Docker work pool, the worker uses a

DockerRegistryCredentials

block to run a

docker login

via the Docker SDK before pulling your image. - Your template has

registry_credentials

but it’s

null

by default — you need to create and attach a credentials block. How to fix Option A (recommended): use a Docker Registry Credentials block 1) Create the block programmatically (example for GCR and Artifact Registry):

from pathlib import Path
from prefect_docker.credentials import DockerRegistryCredentials

# GCR example:
DockerRegistryCredentials(
    username="_json_key",
    password=Path("/path/to/sa.json").read_text(),   # entire JSON key contents
    registry_url="<http://gcr.io|gcr.io>",                           # or <http://us.gcr.io|us.gcr.io>, <http://eu.gcr.io|eu.gcr.io>, <http://asia.gcr.io|asia.gcr.io>
).save("gcr-creds", overwrite=True)

# Artifact Registry example:
DockerRegistryCredentials(
    username="_json_key",
    password=Path("/path/to/sa.json").read_text(),   # entire JSON key contents
    registry_url="us-central1-docker.pkg.dev",       # <region>-docker.pkg.dev
).save("gar-creds", overwrite=True)

2) In your Docker work pool’s job template, set the

registry_credentials

variable to the block you created (in the UI, choose “Docker Registry Credentials” and select your block name — e.g.,

gcr-creds

). 3) Ensure your

image

includes the full registry host: - GCR:

<http://gcr.io/<project>/<image>:<tag|gcr.io/<project>/<image>:<tag>>

(or

<http://us.gcr.io|us.gcr.io>

,

<http://eu.gcr.io|eu.gcr.io>

,

<http://asia.gcr.io|asia.gcr.io>

) - Artifact Registry:

<region>-docker.pkg.dev/<project>/<repo>/<image>:<tag>

Option B: configure Docker on the host If you prefer not to use a Prefect block, authenticate the Docker daemon on the host where the worker runs: - Artifact Registry: Google docs - Container Registry: Google docs Examples: - With a service account JSON key:

docker login -u _json_key -p "$(cat /path/to/sa.json)" <http://gcr.io|gcr.io>
docker login -u _json_key -p "$(cat /path/to/sa.json)" us-central1-docker.pkg.dev

- Or configure the gcloud credential helper:

gcloud auth configure-docker <http://gcr.io|gcr.io>,<http://us.gcr.io|us.gcr.io>,<http://eu.gcr.io|eu.gcr.io>,<http://asia.gcr.io|asia.gcr.io>
gcloud auth configure-docker us-central1-docker.pkg.dev

Common pitfalls to check -

registry_url

must be just the registry hostname (e.g.,

<http://gcr.io|gcr.io>

,

us-central1-docker.pkg.dev

), not the full image path. -

username

should be

_json_key

when using a service account JSON key. -

password

must be the entire JSON file contents, not a path. - Make sure the worker has the Docker integration installed (

prefect[docker]

) since you’re using a Docker work pool. - Verify the image string includes the registry host (e.g.,

<http://gcr.io/|gcr.io/>...

or

<region>-docker.pkg.dev/...

). If you still get “unauthenticated” after setting a credentials block, please share: - Which registry you’re using (GCR vs Artifact Registry) - The exact registry host in your

image

and in

registry_url

- The exact error message snippet I can then spot-check your settings and get you unblocked quickly.

Short answer: almost — but a few important fixes. What’s good - The fields align with DockerWorkerJobConfiguration. - Using network_mode with networks defaulting to [] is fine. - Volumes/env look reasonable. What to fix 1) Remove quotes around non-string variables in job_configuration Right now you have quotes around arrays/objects/booleans and the block. That will coerce them to strings and prevent Prefect from parsing them correctly (including registry_credentials). Use this pattern:

{
  "variables": {
    "properties": {
      "env": { "type": "object", "title": "Environment Variables", "default": {
        "PREFECT_LOCAL_STORAGE_PATH": "/home/prefect/.prefect/storage",
        "GOOGLE_APPLICATION_CREDENTIALS": "/secrets/sa.json",
        "PREFECT_RESULTS_PERSIST_BY_DEFAULT": "true"
      }},
      "name": { "type": "string", "title": "Container Name", "default": "prefect-flow" },
      "image": { "type": "string", "title": "Image" },
      "labels": { "type": "object", "title": "Labels", "default": {} },
      "command": {
        "anyOf": [ { "type": "string" }, { "type": "null" } ],
        "title": "Command", "default": null
      },
      "volumes": {
        "type": "array", "title": "Volumes",
        "default": [
          "prefect-worker-docker_secrets:/secrets:ro",
          "prefect-worker-docker_prefect-cache:/home/prefect/.prefect"
        ]
      },
      "networks": { "type": "array", "title": "Networks", "default": [], "description": "MUST be empty for network_mode to work" },
      "auto_remove": { "type": "boolean", "title": "Auto Remove Container", "default": true },
      "network_mode": { "type": "string", "title": "Network Mode", "default": "container:prefect-worker-docker-tailscale-1" },
      "stream_output": { "type": "boolean", "title": "Stream Output", "default": true },
      "image_pull_policy": {
        "anyOf": [
          { "type": "string", "enum": ["IfNotPresent", "Always", "Never"] },
          { "type": "null" }
        ],
        "title": "Image Pull Policy",
        "default": "IfNotPresent"
      },
      "registry_credentials": {
        "title": "Docker Registry Credentials Block",
        "default": null
      }
    }
  },
  "job_configuration": {
    "env": {{ env }},
    "name": "{{ name }}",
    "image": "{{ image }}",
    "labels": {{ labels }},
    "command": {{ command }},
    "volumes": {{ volumes }},
    "networks": {{ networks }},
    "auto_remove": {{ auto_remove }},
    "network_mode": {{ network_mode }},
    "stream_output": {{ stream_output }},
    "image_pull_policy": {{ image_pull_policy }},
    "registry_credentials": {{ registry_credentials }}
  }
}

2) Ensure you actually attach a DockerRegistryCredentials block - In the UI, set

registry_credentials

to your block. If it stays null, the worker will pull without auth and GCR/AR will return “unauthenticated”. 3) Optional: set image_pull_policy - For private registries, using

"Always"

can be helpful when testing auth. GCR/Artifact Registry specifics - registry_url: use the registry host only (e.g.,

<http://gcr.io|gcr.io>

,

<http://us.gcr.io|us.gcr.io>

,

<http://eu.gcr.io|eu.gcr.io>

, or

<region>-docker.pkg.dev

) - username:

_json_key

- password: the entire JSON contents of your service account key - image must include the registry host, e.g.: - GCR:

<http://gcr.io/<project>/<image>:<tag|gcr.io/<project>/<image>:<tag>>

- AR:

<region>-docker.pkg.dev/<project>/<repo>/<image>:<tag>

How to confirm it’s working - In worker logs you should see “Logging into Docker registry...” before the pull when

registry_credentials

is set. - If you still get “unauthenticated”, double-check the block values and that your

image

host matches

registry_url

. Docs/refs - Docker deployment guide: https://docs.prefect.io/v3/how-to-guides/deployment_infra/docker - Work pools: https://docs.prefect.io/v3/concepts/work-pools - prefect-docker integration: https://docs.prefect.io/integrations/prefect-docker - DockerRegistryCredentials API: https://reference.prefect.io/prefect_docker/credentials/

If you share which registry host you’re targeting and how you set the

registry_credentials

variable in the UI, I can double-check the exact values.

Got it — this usually happens when the variables schema doesn’t include the Block definition/reference the UI needs to render a selector. For the UI to show a “Docker Registry Credentials” picker, your template must: - Define the block in

variables.definitions.DockerRegistryCredentials

- Reference it in

variables.properties.registry_credentials

with a

$ref

- Pass the variable unquoted in

job_configuration

(i.e.,

{{ registry_credentials }}

not

"{{ registry_credentials }}"

) Two easy ways to fix Option A — Start from the default Docker template (recommended) - Run:

prefect work-pool get-default-base-job-template --type docker

- Paste that into your work pool’s Base Job Template in the UI. - Then adjust only the defaults you care about (env, volumes, network_mode, etc.). Leave the

registry_credentials

property and definitions as-is so the UI renders the block selector. Option B — Use this minimal working template with a proper block reference Paste this into your work pool Base Job Template. It keeps your defaults and adds the correct schema for

registry_credentials

so the UI shows the block dropdown. ``` { "variables": { "type": "object", "definitions": { "DockerRegistryCredentials": { "type": "object", "block_type_slug": "docker-registry-credentials", "description": "Store credentials for interacting with a Docker Registry.", "properties": { "username": { "type": "string", "title": "Username", "description": "The username to log into the registry with." }, "password": { "type": "string", "title": "Password", "description": "The password to log into the registry with.", "format": "password", "writeOnly": true }, "registry_url": { "type": "string", "title": "Registry Url", "description": "The URL to the registry. Generally, \"http\" or \"https\" can be omitted.", "examples": ["index.docker.io"] }, "reauth": { "type": "boolean", "title": "Reauth", "description": "Whether or not to reauthenticate on each interaction.", "default": true } }, "required": ["username", "password", "registry_url"] } }, "properties": { "env": { "type": "object", "title": "Environment Variables", "default": { "PREFECT_LOCAL_STORAGE_PATH": "/home/prefect/.prefect/storage", "GOOGLE_APPLICATION_CREDENTIALS": "/secrets/sa.json", "PREFECT_RESULTS_PERSIST_BY_DEFAULT": "true" } }, "name": { "type": "string", "title": "Container Name", "default": "prefect-flow" }, "image": { "type": "string", "title": "Image" }, "labels": { "type": "object", "title": "Labels", "default": {} }, "command": { "anyOf": [{ "type": "string" }, { "type": "null" }], "title": "Command", "default": null }, "volumes": { "type": "array", "title": "Volumes", "default": [ "prefect-worker-docker_secrets/secretsro", "prefect-worker-docker_prefect-cache:/home/prefect/.prefect" ] }, "networks": { "type": "array", "title": "Networks", "default": [], "description": "MUST be empty for network_mode to work" }, "auto_remove": { "type": "boolean", "title": "Auto Remove Container", "default": true }, "network_mode": { "type": "string", "title": "Network Mode", "default": "container:prefect-worker-docker-tailscale-1" }, "stream_output": { "type": "boolean", "title": "Stream Output", "default": true }, "image_pull_policy": { "anyOf": [ { "type": "string", "enum": ["IfNotPresent", "Always", "Never"] },

{ "type": "null" } ], "title": "Image Pull Policy", "default": "IfNotPresent" }, "registry_credentials": { "title": "Docker Registry Credentials", "anyOf": [ { "$ref": "#/definitions/DockerRegistryCredentials" }, { "type": "null" } ], "default": null, "description": "Credentials for logging into a Docker registry to pull images from." } } }, "job_configuration": { "env": {{ env }}, "name": "{{ name }}", "image": "{{ image }}", "labels": {{ labels }}, "command": {{ command }}, "volumes": {{ volumes }}, "networks": {{ networks }}, "auto_remove": {{ auto_remove }}, "network_mode": {{ network_mode }}, "stream_output": {{ stream_output }}, "image_pull_policy": {{ image_pull_policy }}, "registry_credentials": {{ registry_credentials }} } }

After saving, when you create or edit a deployment targeting this work pool, you should see:
- A “Docker Registry Credentials” picker (to select/create your `DockerRegistryCredentials` block)
- Your default env/volumes/network_mode populated

If you prefer to hardcode a specific block (no UI selector), you can set:

"registry_credentials": "{{ prefect.blocks.docker-registry-credentials.gcr-creds }}" ``` in

job_configuration

, and remove the

registry_credentials

variable from

variables.properties

. Docs for reference - Docker deployment guide: https://docs.prefect.io/v3/how-to-guides/deployment_infra/docker - Work pools: https://docs.prefect.io/v3/concepts/work-pools - prefect-docker integration: https://docs.prefect.io/integrations/prefect-docker - DockerRegistryCredentials API: https://reference.prefect.io/prefect_docker/credentials/ If the UI still doesn’t render the field, can you paste a redacted copy of the template you saved in the work pool? I’ll spot the missing schema piece.

Thanks for the heads up — and sorry for the confusion earlier. I gave you one bad steer: in Prefect 3 base job templates, placeholders must remain quoted as strings (e.g., "{{ var }}"). The engine injects JSON at render time. Unquoting them can cause the UI/API to fail with the generic “Error updating deployment”. What to do now - Reset your work pool template to something known-good using the default Docker template, then add your defaults. - Keep all placeholders in job_configuration quoted. Known-good template you can paste This template keeps your defaults and renders the Docker Registry Credentials picker correctly. Note the quotes in job_configuration. ``` { "variables": { "type": "object", "definitions": { "DockerRegistryCredentials": { "type": "object", "block_type_slug": "docker-registry-credentials", "description": "Store credentials for interacting with a Docker Registry.", "properties": { "username": { "type": "string", "title": "Username", "description": "The username to log into the registry with." }, "password": { "type": "string", "title": "Password", "description": "The password to log into the registry with.", "format": "password", "writeOnly": true }, "registry_url": { "type": "string", "title": "Registry Url", "description": "The URL to the registry. Generally, \"http\" or \"https\" can be omitted.", "examples": ["index.docker.io"] }, "reauth": { "type": "boolean", "title": "Reauth", "description": "Whether or not to reauthenticate on each interaction.", "default": true } }, "required": ["username", "password", "registry_url"] } }, "properties": { "env": { "type": "object", "title": "Environment Variables", "default": { "PREFECT_LOCAL_STORAGE_PATH": "/home/prefect/.prefect/storage", "GOOGLE_APPLICATION_CREDENTIALS": "/secrets/sa.json", "PREFECT_RESULTS_PERSIST_BY_DEFAULT": "true" } }, "name": { "type": "string", "title": "Container Name", "default": "prefect-flow" }, "image": { "type": "string", "title": "Image" }, "labels": { "type": "object", "title": "Labels", "default": {} }, "command": { "anyOf": [{ "type": "string" }, { "type": "null" }], "title": "Command", "default": null }, "volumes": { "type": "array", "title": "Volumes", "default": [ "prefect-worker-docker_secrets/secretsro", "prefect-worker-docker_prefect-cache:/home/prefect/.prefect" ] }, "networks": { "type": "array", "title": "Networks", "default": [], "description": "MUST be empty for network_mode to work" }, "auto_remove": { "type": "boolean", "title": "Auto Remove Container", "default": true }, "network_mode": { "type": "string", "title": "Network Mode", "default": "container:prefect-worker-docker-tailscale-1" }, "stream_output": { "type": "boolean", "title": "Stream Output", "default": true }, "image_pull_policy": { "anyOf": [ { "type": "string", "enum": ["IfNotPresent", "Always", "Never"] }, { "type": "null" } ], "title": "Image Pull Policy", "default": "IfNotPresent" }, "registry_credentials": { "title": "Docker Registry Credentials", "anyOf": [ { "$ref": "#/definitions/DockerRegistryCredentials" }, { "type": "null" } ], "default": null, "description": "Credentials for logging into a Docker registry to pull images from." },

"mem_limit": { "anyOf": [{ "type": "string" }, { "type": "null" }], "title": "Memory Limit", "default": null }, "memswap_limit": { "anyOf": [{ "type": "string" }, { "type": "null" }], "title": "Memory Swap Limit", "default": null }, "privileged": { "type": "boolean", "title": "Privileged", "default": false }, "container_create_kwargs": { "anyOf": [{ "type": "object" }, { "type": "null" }], "title": "Container Create Kwargs", "default": null } } }, "job_configuration": { "command": "{{ command }}", "env": "{{ env }}", "labels": "{{ labels }}", "name": "{{ name }}", "image": "{{ image }}", "registry_credentials": "{{ registry_credentials }}", "image_pull_policy": "{{ image_pull_policy }}", "networks": "{{ networks }}", "network_mode": "{{ network_mode }}", "auto_remove": "{{ auto_remove }}", "volumes": "{{ volumes }}", "stream_output": "{{ stream_output }}", "mem_limit": "{{ mem_limit }}", "memswap_limit": "{{ memswap_limit }}", "privileged": "{{ privileged }}", "container_create_kwargs": "{{ container_create_kwargs }}" } }

If you still get “Error updating deployment”
- Check the work pool template via CLI to ensure it saved as valid JSON:
  - Inspect: `prefect work-pool inspect "<your-pool-name>"`
  - Or reapply via file to get better error messages:
    - Save the JSON to `template.json`
    - Run: `prefect work-pool update "<your-pool-name>" --base-job-template template.json`
- In the UI, open the browser DevTools network tab; the failing request usually includes a detailed validation message from the API. Paste that back here and I’ll pinpoint the issue.

Hardcode as a fallback
If you just want to unblock pulls immediately, you can hardcode a specific block reference and skip the UI selector:
- Create the block `docker-registry-credentials.gcr-creds`
- Set in job_configuration:

"registry_credentials": "{{ prefect.blocks.docker-registry-credentials.gcr-creds }}" ``` - Remove the

registry_credentials

entry from

variables.properties

so the UI doesn’t try to render it. Docs/refs - Default Docker template retrieval:

prefect work-pool get-default-base-job-template --type docker

- Docker guide: https://docs.prefect.io/v3/how-to-guides/deployment_infra/docker - prefect-docker integration: https://docs.prefect.io/integrations/prefect-docker - DockerRegistryCredentials API: https://reference.prefect.io/prefect_docker/credentials/ If you can share the exact API error text (from the network response), I’ll nail down the fix quickly.

Thank you for grabbing that server traceback — it pinpoints the issue. Root cause - Prefect’s server preprocesses the base job template’s variables schema and expects each entry under

variables.definitions

to include a string

title

. - In our earlier template, the

DockerRegistryCredentials

definition was missing

"title"

, which triggers that exact failure when the server runs: schema["definitions"][definition["title"]] = { ... } Fix Add a

title

to the DockerRegistryCredentials definition and ensure the ref matches the definition key. Here’s a corrected, known-good template you can paste into the work pool Base Job Template. It: - Includes the required

title

in definitions - Keeps placeholders quoted in job_configuration - Renders the Docker Registry Credentials picker in the UI ``` { "variables": { "type": "object", "definitions": { "DockerRegistryCredentials": { "title": "DockerRegistryCredentials", "type": "object", "block_type_slug": "docker-registry-credentials", "description": "Store credentials for interacting with a Docker Registry.", "properties": { "username": { "type": "string", "title": "Username", "description": "The username to log into the registry with." }, "password": { "type": "string", "title": "Password", "description": "The password to log into the registry with.", "format": "password", "writeOnly": true }, "registry_url": { "type": "string", "title": "Registry Url", "description": "The URL to the registry. Generally, \"http\" or \"https\" can be omitted.", "examples": ["index.docker.io"] }, "reauth": { "type": "boolean", "title": "Reauth", "description": "Whether or not to reauthenticate on each interaction.", "default": true } }, "required": ["username", "password", "registry_url"] } }, "properties": { "env": { "type": "object", "title": "Environment Variables", "default": { "PREFECT_LOCAL_STORAGE_PATH": "/home/prefect/.prefect/storage", "GOOGLE_APPLICATION_CREDENTIALS": "/secrets/sa.json", "PREFECT_RESULTS_PERSIST_BY_DEFAULT": "true" } }, "name": { "type": "string", "title": "Container Name", "default": "prefect-flow" }, "image": { "type": "string", "title": "Image" }, "labels": { "type": "object", "title": "Labels", "default": {} }, "command": { "anyOf": [{ "type": "string" }, { "type": "null" }], "title": "Command", "default": null }, "volumes": { "type": "array", "title": "Volumes", "default": [ "prefect-worker-docker_secrets/secretsro", "prefect-worker-docker_prefect-cache:/home/prefect/.prefect" ] }, "networks": { "type": "array", "title": "Networks", "default": [], "description": "MUST be empty for network_mode to work" }, "auto_remove": { "type": "boolean", "title": "Auto Remove Container", "default": true }, "network_mode": { "type": "string", "title": "Network Mode", "default": "container:prefect-worker-docker-tailscale-1" }, "stream_output": { "type": "boolean", "title": "Stream Output", "default": true }, "image_pull_policy": { "anyOf": [ { "type": "string", "enum": ["IfNotPresent", "Always", "Never"] }, { "type": "null" } ], "title": "Image Pull Policy", "default": "IfNotPresent" }, "registry_credentials": { "title": "Docker Registry Credentials", "anyOf": [ { "$ref": "#/definitions/DockerRegistryCredentials" }, { "type": "null" }

], "default": null, "description": "Credentials for logging into a Docker registry to pull images from." }, "mem_limit": { "anyOf": [{ "type": "string" }, { "type": "null" }], "title": "Memory Limit", "default": null }, "memswap_limit": { "anyOf": [{ "type": "string" }, { "type": "null" }], "title": "Memory Swap Limit", "default": null }, "privileged": { "type": "boolean", "title": "Privileged", "default": false }, "container_create_kwargs": { "anyOf": [{ "type": "object" }, { "type": "null" }], "title": "Container Create Kwargs", "default": null } } }, "job_configuration": { "command": "{{ command }}", "env": "{{ env }}", "labels": "{{ labels }}", "name": "{{ name }}", "image": "{{ image }}", "registry_credentials": "{{ registry_credentials }}", "image_pull_policy": "{{ image_pull_policy }}", "networks": "{{ networks }}", "network_mode": "{{ network_mode }}", "auto_remove": "{{ auto_remove }}", "volumes": "{{ volumes }}", "stream_output": "{{ stream_output }}", "mem_limit": "{{ mem_limit }}", "memswap_limit": "{{ memswap_limit }}", "privileged": "{{ privileged }}", "container_create_kwargs": "{{ container_create_kwargs }}" } }

Why this works
- The server’s schema preprocessor specifically looks for `variables.definitions[...].title` when a definition has `block_type_slug` and rewrites it for block references. Without `title`, it throws the exception you saw.

If you still hit an error after pasting this:
- Try updating the pool from a file to get a clearer error:
  - Save as template.json
  - Run:

prefect work-pool update "<your-pool-name>" --base-job-template template.json ``` - If it fails, please paste the CLI error output or the server log right after the call — that will include the precise validation message. Once this saves, the UI should let you select your Docker Registry Credentials block and your worker will be able to pull from GCR/Artifact Registry.

Almost — two small but important corrections: - Username should be exactly

_json_key

(single underscores as shown). Not

___json__key

. - Registry URL should be just the hostname, no scheme:

europe-west3-docker.pkg.dev

(not

http://...

and not Slack link markup). And yes, the password should be the entire contents of your GCP service account JSON key. Example: create the block

from pathlib import Path
from prefect_docker.credentials import DockerRegistryCredentials

DockerRegistryCredentials(
    username="_json_key",
    password=Path("/path/to/service-account.json").read_text(),  # full JSON content
    registry_url="europe-west3-docker.pkg.dev",                  # hostname only
    reauth=True,
).save("gar-ew3-creds", overwrite=True)

Then in your work pool’s job template, select that block for

registry_credentials

(or hardcode it):

"registry_credentials": "{{ prefect.blocks.docker-registry-credentials.gar-ew3-creds }}"

Also double-check: - Your image includes the same host:

europe-west3-docker.pkg.dev/<project-id>/<repo>/<image>:<tag>

- The service account has at least Artifact Registry Reader on the repo/project - Worker logs show “Logging into Docker registry...” before pulling Google’s doc for this method: Authenticate to Artifact Registry with a JSON key If you still get an auth error after this, please share: - The exact image string you’re pulling - The worker log lines around “Logging into Docker registry…” - Whether you configured the block via UI or Python (and its name)

Short answer: correct — with

image_pull_policy: "IfNotPresent"

, the Docker worker will NOT check the registry for updates if the image tag already exists locally. It will reuse the local image, even if you’ve pushed a newer image under the same tag. Ways to get the newest image: - Set

image_pull_policy

to

"Always"

for that deployment/work pool so the worker pulls before each run. - Use unique/immutable tags per build (e.g.,

my-image:gitsha

,

my-image:2025-01-01-1234

) so the worker sees a “new” tag and pulls it. - Manually update the image on the worker host: run

docker pull <your-image>

where the worker runs. Subsequent runs with

IfNotPresent

will use the updated local image. - Remove the local image on the worker host (

docker rmi <your-image>

) so the worker has to pull it again. Tip: For development,

"Always"

is convenient. For production, prefer immutable tags or digests (e.g.,

...@sha256:<digest>

) to guarantee the exact image that runs. Docs: Docker deployment guide

Likely causes, given your setup with a Docker work pool: - You’re running an old image - If

image_pull_policy

is

IfNotPresent

and the tag already exists locally, the worker won’t pull your updated image. You’ll keep running the old image that’s missing the new deps. - Fix: temporarily set

image_pull_policy

to

Always

, or use unique/immutable tags per build (e.g.,

my-image:gitsha

) and update the deployment to that exact tag. You can also manually

docker pull <image>

or

docker rmi <image>

on the worker host. - The deployment points to the wrong tag - Make sure the deployment’s “Image” matches the image you actually built and pushed. Check worker logs for “Creating docker container from image '...'” to confirm what it’s running. - Dependencies aren’t in the image - Your Dockerfile may not install them (or installs them in a venv not on PATH). Verify by running locally:

docker run --rm <your-image> python -c "import <package_or_module>"

- If you use a venv in the image, ensure you set PATH, e.g.:

RUN python -m venv /venv
    ENV PATH="/venv/bin:$PATH"

- If you need OS-level libs (e.g., for psycopg2, snowflake, etc.), add

apt-get install

lines. - Your project code isn’t installed in the image - If your flow imports your own package/modules, you must copy and install them in the image. - Example Dockerfile:

FROM prefecthq/prefect:3-python3.12
    WORKDIR /opt/prefect

    # OS deps if needed
    # RUN apt-get update && apt-get install -y build-essential libpq-dev && rm -rf /var/lib/apt/lists/*

    # Install Python deps first for better caching
    COPY requirements.txt .
    RUN pip install --no-cache-dir -r requirements.txt

    # Copy your code and install it (if it’s a package)
    COPY . .
    # RUN pip install --no-cache-dir .

- If you use pyproject/poetry, either export a requirements.txt or do

pip install .

after copying pyproject.toml (and lockfile if you have one). - Missing Prefect integrations - If you import integrations like

prefect_gcp

, install them in the image:

pip install prefect-gcp

- Same for others (aws, docker, etc.). Quick checklist to unblock 1) Force a fresh image to run - Rebuild with a unique tag, push it, update your deployment to that exact tag, and set

image_pull_policy

to

Always

for a run or two. 2) Sanity-check the image - Locally:

docker run --rm <image> python -c "import <thing-that-fails>"

3) Verify deployment/worker are using the expected image - Check the deployment’s Image string in the UI and the worker log message that shows the image it’s using. If you can share: - The exact ModuleNotFoundError/ImportError text - The image reference your deployment is using - Your Dockerfile (redacted as needed) - Whether you’re using a custom

prefect.yaml

with a docker build step or building manually …I can pinpoint the issue and give you a targeted fix. Docker guide for reference: https://docs.prefect.io/v3/how-to-guides/deployment_infra/docker

Most “missing dependency” crashes I see with this pattern come from one of these: 1) You’re running an old image - If your work pool/deployment is set to IfNotPresent and the tag exists locally, the worker will reuse the old image (with old deps). - Fix: temporarily set image_pull_policy to "Always" or use a new, unique tag for each build and update the deployment to that exact tag. 2) Your project/package isn’t installed into the image - Copying the code doesn’t make it importable if you use a “src/” layout or import your package by name. - Fix one of: - Install your package:

# after COPY and setting WORKDIR
    RUN pip install --no-cache-dir .

- Or, add your source dir to PYTHONPATH:

ENV PYTHONPATH="/opt/prefect/redacted/src:${PYTHONPATH}"

- Quick sanity check:

docker run --rm -it <your-image> python -c "import your_package"

If this fails, it’s not installed or not on sys.path. 3) A required runtime dependency isn’t in your export - You’re doing

uv export --frozen --no-dev

. If you accidentally categorized runtime deps as “dev” in your pyproject, they won’t be exported and pip won’t install them. - Fix: either move those deps to the main/runtime group or temporarily remove

--no-dev

to confirm the hypothesis. 4) System packages needed to build/install some wheels are missing - If you have packages like psycopg2, uvicorn with extras, cryptography, etc., you may need build tools/headers. - Add what you need, for example:

RUN apt-get update && apt-get install -y --no-install-recommends \
      build-essential libpq-dev gcc \
    && rm -rf /var/lib/apt/lists/*

5) Private GitHub dependencies not actually installing - Your

git config

rewrite is clever, but be aware: - ARG GITHUB_TOKEN at build time is embedded in build history layers; prefer BuildKit secrets for tokens if possible. - Make sure your requirements include git+https URLs so pip uses git (and thus your rewrite). - If in doubt, test inside the container:

docker run --rm -it <image> bash -lc "cat requirements.txt; pip install -r requirements.txt"

Suggested tightened Dockerfile - Installs system deps first, installs Python deps from lock, then installs your package so imports work.

FROM prefecthq/prefect:3-python3.12

# Optional: system build deps if your wheels need compiling
# RUN apt-get update && apt-get install -y --no-install-recommends \
#     build-essential libpq-dev git \
#   && rm -rf /var/lib/apt/lists/*

# Install uv and git (if not using the commented block above)
RUN pip install --no-cache-dir uv && apt-get update && apt-get install -y --no-install-recommends git && rm -rf /var/lib/apt/lists/*

ARG GITHUB_TOKEN

# Copy project files
COPY . /opt/prefect/redacted/
WORKDIR /opt/prefect/redacted/

# Resolve and install runtime deps from lock
RUN git config --global url."https://${GITHUB_TOKEN}@github.com/".insteadOf "<https://github.com/>" && \
    uv export --frozen --no-dev -o requirements.txt && \
    pip install --no-cache-dir -r requirements.txt && \
    git config --global --unset url."https://${GITHUB_TOKEN}@github.com/".insteadOf || true

# Ensure your project is importable
# If you have a pyproject/SETUP and a src/ layout, this makes imports succeed
RUN pip install --no-cache-dir .

# Or alternatively:
# ENV PYTHONPATH="/opt/prefect/redacted/src:${PYTHONPATH}"

Security note about GITHUB_TOKEN - Using ARG embeds the token in the build history. Prefer Docker BuildKit secrets if you can: - Add at top of Dockerfile:

# syntax=docker/dockerfile:1.6

- Then: ``` RUN --mount=type=secret,id=GH_TOKEN \ GITHUB_TOKEN=$(cat /run/secrets/GH_TOKEN) && \ git config --global url."https://${GITHUB_TOKEN}@github.com/".insteadOf "https://github.com/" && \ uv export --frozen --no-dev -o requirements.txt && \ pip install -r requirements.txt && \

git config --global --unset url."https://${GITHUB_TOKEN}@github.com/".insteadOf || true

- Build with: `docker build --secret id=GH_TOKEN,env=GITHUB_TOKEN .`

Two quick validations
- Verify the image contains what you expect:

docker run --rm <image> python -c "import sys; print(sys.version); import your_pkg, some_dep" ``` - Check the worker is using that image: - In logs: “Creating docker container from image '...'” - If not, update the deployment image tag and set image_pull_policy to "Always" for a run. If you can share the exact ModuleNotFoundError and your repo layout (e.g., do you have a src/ directory?), I can tailor the fix precisely.

There are three places “cache” can bite you during a Docker-based deploy. Here’s how to force fresh artifacts at each step: 1) Build: don’t use Docker layer cache - If you’re using prefect.yaml build steps, set

ignore_cache: true

on the build step. You can also force pulling newer base images with

build_kwargs: { pull: true }

.

# prefect.yaml
deployments:
- name: my-deploy
  work_pool:
    name: my-docker-pool
  build:
  - prefect_docker.deployments.steps.build_docker_image:
      image_name: europe-west3-docker.pkg.dev/<project>/<repo>/<image>
      tag: "{{ git_sha }}"
      dockerfile: Dockerfile
      ignore_cache: true
      build_kwargs:
        pull: true
  push:
  - prefect_docker.deployments.steps.push_docker_image:
      image_name: europe-west3-docker.pkg.dev/<project>/<repo>/<image>
      tag: "{{ git_sha }}"

- If you build manually, use:

docker build --no-cache --pull -t <image:tag> .

Refs: prefect-docker integration, Docker deployment guide. The build step signature (shows

ignore_cache

and

**build_kwargs

): build_docker_image reference. 2) Runtime pull: don’t reuse a local image - In your Docker work pool/job variables, set

image_pull_policy

to

"Always"

so the worker pulls the image before each run. - Alternatively, use immutable/unique tags (e.g.,

:gitsha

) so

IfNotPresent

still pulls a “new” tag. 3) Tagging: guarantee you’re running the new build - Always push with a unique tag and point the deployment at that exact tag. - Verify in worker logs: “Creating docker container from image '…:your-tag'”. If you want, share your current prefect.yaml deployment section and I’ll mark exactly where to add

ignore_cache

and

image_pull_policy

.

You’re hitting two different caches. To force a truly fresh build, disable both: - Prefect step cache: add

ignore_cache: true

- Docker layer cache: add

nocache: true

(passed through to the Docker SDK build) In prefect.yaml

deployments:
- name: my-deploy
  work_pool:
    name: my-docker-pool
  build:
  - prefect_docker.deployments.steps.build_docker_image:
      image_name: europe-west3-docker.pkg.dev/<project>/<repo>/<image>
      tag: "{{ git_sha }}"
      dockerfile: Dockerfile
      ignore_cache: true        # bypass Prefect step cache
      nocache: true             # bypass Docker layer cache (like --no-cache)
      build_kwargs:
        pull: true              # also repull the base image
  push:
  - prefect_docker.deployments.steps.push_docker_image:
      image_name: europe-west3-docker.pkg.dev/<project>/<repo>/<image>
      tag: "{{ git_sha }}"

If you’re building manually

docker build --no-cache --pull -t <image:tag> .

Extra hard reset (if Docker still reuses layers) - Prune build cache on the builder host: - Classic builder:

docker builder prune -af

- Buildx builder:

docker buildx prune -af

- Or remove any cache sources you’re referencing via

cache_from

. At runtime, ensure the worker actually pulls your new image - Set your deployment’s

image

to the unique tag you just built (e.g., a git SHA) - Set

image_pull_policy

to

"Always"

for a run or two, then revert to

"IfNotPresent"

if desired Why your earlier attempts still used cache -

ignore_cache

only disables the Prefect step’s in-memory cache; without

nocache: true

, the Docker daemon can still serve cached layers. - Reusing the same tag with

IfNotPresent

can cause the worker to skip pulling the updated image if it’s already present locally. If you paste your current build section from prefect.yaml, I can confirm the exact placement of

ignore_cache

and

nocache

.

You’re close, but there are a few gotchas that can still make the build look “cached.” Here’s how to force a truly fresh build and verify it. What to change in your prefect.yaml - Keep ignore_cache: true (skips Prefect’s step cache) - Add nocache: true AND pull: true to the build kwargs (skip Docker layer cache and repull base image) - Optionally add forcerm: true to clean up intermediate containers - Add a cache-busting build arg that changes every run (e.g., timestamp or git SHA) and consume it in your Dockerfile with ARG CACHE_BUST Example

build:
- prefect_docker.deployments.steps.build_docker_image:
    id: build-image
    requires: prefect-docker>=0.4.0
    image_name: redacted
    tag: "{{ git_sha }}"
    dockerfile: "Dockerfile"
    platform: "linux/amd64"
    ignore_cache: true       # bypass Prefect step cache
    nocache: true            # bypass Docker layer cache
    pull: true               # always repull base image
    forcerm: true            # remove intermediate containers
    buildargs:
      GITHUB_TOKEN: "{{ $GITHUB_TOKEN }}"
      CACHE_BUST: "{{ now }}"

Update your Dockerfile to use the cache-buster

FROM prefecthq/prefect:3-python3.12

# Consume cache-busting arg (ensures following layers rebuild)
ARG CACHE_BUST

RUN pip install --no-cache-dir uv && apt-get update && apt-get install -y --no-install-recommends git && rm -rf /var/lib/apt/lists/*

ARG GITHUB_TOKEN
COPY . /opt/prefect/redacted/
WORKDIR /opt/prefect/redacted/

RUN git config --global url."https://${GITHUB_TOKEN}@github.com/".insteadOf "<https://github.com/>" && \
    uv export --frozen --no-dev -o requirements.txt && \
    pip install --no-cache-dir -r requirements.txt && \
    git config --global --unset url."https://${GITHUB_TOKEN}@github.com/".insteadOf || true

# Ensure your package is importable, if needed
# RUN pip install --no-cache-dir .

Also make sure at runtime - Use a unique tag (you already use {{ git_sha }}) and push it - Point the deployment image at that exact tag - Set image_pull_policy to "Always" for a run or two so the worker pulls the new tag If it still appears cached - Prune the builder cache on the build machine: - Classic builder:

docker builder prune -af

- Buildx builder:

docker buildx prune -af

(if you’re using buildx) - Run a manual build to compare behavior:

docker build --no-cache --pull \
    --platform linux/amd64 \
    --build-arg GITHUB_TOKEN=... \
    --build-arg CACHE_BUST=$(date +%s) \
    -t redacted:manual-test .

If this is fresh but Prefect isn’t, we’ll debug the step invocation. Extra verification - Turn on debug logs: set

PREFECT_LOGGING_LEVEL=DEBUG

before running `prefect deploy`; you should see logs indicating the

@cacheable

decorator was ignored and that the Docker API build was invoked with nocache/pull. - Confirm the worker is actually running the new tag in its logs: “Creating docker container from image 'redacted:…'” If you paste the build logs that still show “CACHED” lines, I’ll point out which layer is being reused and why.

Short answer: Prefect is retrying because something is telling it to. The Docker work pool doesn’t add its own retry logic; the server will reschedule runs when retries are configured or an Automation is re-running your deployment. Check these common causes: - Flow/task decorators: Look for

@flow(retries=...)

or

@task(retries=...)

(including

retry_delay_seconds

or

exponential_backoff

). Any >0 will auto-reschedule on failure. - Global defaults: The worker/deployment environment may set -

PREFECT_FLOW_DEFAULT_RETRIES

-

PREFECT_FLOW_DEFAULT_RETRY_DELAY_SECONDS

-

PREFECT_TASK_DEFAULT_RETRIES

-

PREFECT_TASK_DEFAULT_RETRY_DELAY_SECONDS

Run

prefect config view

on the worker host to confirm. - Automations: If you created an Automation like “on flow run failed → Run deployment,” it will re-trigger runs on every failure (looks like infinite retries). - You clicked “Retry” in the UI: That schedules another attempt for that specific run. How to stop it: - Remove or set

retries=0

on your `@flow`/`@task` decorators. - Unset global defaults in the worker/deployment environment (or profile) if they’re set. - Disable any Automation that runs the deployment on failure/crash. - If you only want fresh images to run (not related to retries), set

image_pull_policy

to

"Always"

. If you share a snippet of your flow/task decorators and any Automations you’ve set up, plus the output of

prefect config view

from the worker, I can pinpoint the exact source.

You’ve got a few options: From the CLI - Run:

prefect version

This prints the Prefect app version plus details (API version, Python, OS, etc.). - Or the short form:

prefect --version

From the server HTTP API - Application version:

curl <http://YOUR_SERVER/api/admin/version>

Returns the Prefect server application version (e.g., 3.x.y). - API protocol version:

curl <http://YOUR_SERVER/api/version>

Returns the API version (e.g., 0.8.x), which is different from the app version. From the UI - Open the Settings page in the Prefect UI; the version is shown at the top of the page.

Can you clarify what “DAG not loading” means for you? - Is it the Flow Run graph tab in the UI showing a spinner/blank? - Or do runs fail to start (so there’s no graph at all)? - Any specific error in the browser devtools Network/Console or in your server logs? Common causes and quick checks - Very large graphs / heavy mapping - Large fan-out (e.g., thousands of mapped tasks) can make the graph endpoint slow or time out. - Try a smaller test run, chunk your mapping, or split into subflows to see if the graph renders. - Old versions - Upgrade both your server/worker and UI to the latest Prefect 3.x; we’ve fixed multiple graph rendering issues over time. - Check versions: - CLI:

prefect version

- Server HTTP:

curl <http://YOUR_SERVER/api/admin/version>

and

curl <http://YOUR_SERVER/api/version>

- Browser/network issues - Open devtools → Network. Look for failures on: -

/api/flow-runs/{id}/graph

-

/api/task-runs?...flow_run_id=...

- If you’re behind a proxy (NGINX/Cloudflare/ingress), increase timeouts and ensure websockets/long requests are allowed (proxy_read_timeout, keepalive, Upgrade/Connection headers). - Temporarily disable ad blockers/extensions and try a hard refresh. - Flow run crashes early - If the flow crashes before tasks are scheduled, the graph may be empty. - Check the flow run logs for early exceptions (import errors, missing code in the image, bad entrypoint). - Mismatched deployment/image - If your worker can’t load your flow (wrong image/tag, missing dependencies), there may be no DAG to render. - Confirm the worker logs: “Creating docker container from image '...’” and that the container starts without immediate errors. How to get me the right signal - A screenshot or copy of the failing request from devtools Network (status and response body) - The exact flow run ID and a snippet from your server logs around the time you open the graph - Approximate task count in your run (order of magnitude) - Whether you’re behind a reverse proxy and which one (NGINX, Traefik, Cloudflare, etc.) With one of those, I can pinpoint whether this is a size/timeouts issue, a proxy/UI request problem, or a flow/worker failure before the DAG is created.

Short answer: yes, you can use multiple CPUs with a Docker worker, and you can use Dask. You have a few options depending on whether you want parallelism inside a single container or across multiple containers. Option A — Use multiple CPUs within the flow’s container - By default, a container can use all host CPUs unless you limit it. To explicitly control CPU allocation for the job container, set Docker create options via

container_create_kwargs

in your Docker work pool job variables/template:

# In your deployment's job variables or base job template defaults
"container_create_kwargs": {
  "nano_cpus": 4000000000,      # ~4 CPUs (2e9 = 2 CPUs)
  "cpuset_cpus": "0-3"          # optional: pin to specific cores
}

- Then use a multi-process task runner to actually leverage those CPUs inside the container:

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

@task
def cpu_bound(x): 
    # heavy CPU work
    return x * x

@flow(task_runner=ProcessPoolTaskRunner(max_workers=4))
def my_flow():
    futures = [cpu_bound.submit(i) for i in range(100)]
    return [f.result() for f in futures]

Notes: - Use ProcessPoolTaskRunner (or Dask with processes) for CPU-bound work; threads won’t bypass the GIL for CPU-heavy code. - You can also set

mem_limit

in job variables if you need memory control. Option B — Use Dask in a single container (LocalCluster) - Run a Dask LocalCluster inside the container to use multiple processes/threads:

pip install prefect-dask dask[distributed]

from dask.distributed import LocalCluster
from prefect import flow, task
from prefect_dask.task_runners import DaskTaskRunner

@task
def cpu_bound(x): 
    return x * x

@flow(
    task_runner=DaskTaskRunner(
        cluster_class=LocalCluster,
        cluster_kwargs={"n_workers": 4, "threads_per_worker": 1, "processes": True},
    )
)
def my_flow():
    futures = [cpu_bound.submit(i) for i in range(1000)]
    return [f.result() for f in futures]

- Combine with

container_create_kwargs

CPU settings if you want to pin/limit cores. Option C — Use a distributed Dask cluster (multi-container) - Yes, you can point your flow to an existing Dask cluster (e.g., on K8s, Docker Compose, EC2, etc.):

from prefect_dask.task_runners import DaskTaskRunner

@flow(task_runner=DaskTaskRunner(address="<tcp://your-scheduler:8786>"))
def my_flow():
    ...

- The Docker worker just runs your flow container; the flow connects to your external Dask scheduler and distributes work to its workers. - You can also programmatically create a cluster (e.g., dask_kubernetes.KubeCluster) if your runtime has the right permissions/networking. Which should you pick? - Single machine, one container: ProcessPoolTaskRunner or Dask LocalCluster + CPU limits via

container_create_kwargs

. - Multiple machines/containers: Stand up a Dask cluster and connect to it with

DaskTaskRunner(address=...)

. References - Dask task runner for Prefect: https://docs.prefect.io/integrations/prefect-dask - Dask LocalCluster: https://docs.dask.org/en/stable/deploying-python.html#localcluster - Docker CPU constraints: https://docs.docker.com/config/containers/resource_constraints/ - Docker SDK container create options (cpu_shares, cpu_quota, nano_cpus, cpuset_cpus): https://docker-py.readthedocs.io/en/stable/containers.html If you share whether you want single-container multi-core or a distributed cluster, and your environment (single host vs K8s), I can give a concrete config snippet for your work pool template and flow.

Great question. Use DaskTaskRunner when you want Prefect to execute tasks on a Dask scheduler for parallelism that goes beyond a single Python process or machine. Good fits for DaskTaskRunner - You need to fan out a lot of work - Many independent tasks (embarrassingly parallel), e.g., processing hundreds/thousands of items - You want to scale beyond one machine - Connect to a Dask cluster (K8s, ECS, Ray-on-Dask, etc.) and distribute tasks across many workers - You need robust multi-process concurrency - CPU-bound Python workloads (avoid the GIL), or a mix of IO and CPU with flexible worker configs - You want cluster-level elasticity - Dask can adapt worker counts dynamically; Prefect happily schedules into it - You already have a Dask ecosystem - Prefer to keep scheduling/monitoring in Dask while using Prefect for orchestration, retries, logging, schedules, etc. When a simpler runner is better - Small flows or few concurrent tasks - ThreadPoolTaskRunner (IO-bound) or ProcessPoolTaskRunner (CPU-bound) on one machine is lower overhead - Tasks are very short-lived (milliseconds/low seconds) - Dask scheduling/serialization overhead can dominate - Large in-memory payloads between tasks - Pickling and network transfer can be expensive; instead, write to shared storage (S3/GCS/local FS) and pass references Key considerations - Install dependencies in your image/env:

pip install prefect-dask "dask[distributed]"

- Data transfer: avoid passing huge objects between tasks; store-and-reference instead - Serialization: task inputs/outputs must be pickleable - Networking: if using an external Dask cluster, your Docker worker container must reach the scheduler’s address/ports Common patterns - Single host, multi-core via LocalCluster

from dask.distributed import LocalCluster
from prefect import flow, task
from prefect_dask.task_runners import DaskTaskRunner

@task
def work(x): return x * x

@flow(task_runner=DaskTaskRunner(cluster_class=LocalCluster,
                                 cluster_kwargs={"n_workers": 4, "threads_per_worker": 1, "processes": True}))
def my_flow():
    futs = [work.submit(i) for i in range(1000)]
    return [f.result() for f in futs]

- Distributed cluster (existing scheduler)

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

@task
def work(x): return x * x

@flow(task_runner=DaskTaskRunner(address="<tcp://dask-scheduler:8786>"))
def my_flow():
    futs = [work.submit(i) for i in range(1000)]
    return [f.result() for f in futs]

With a Docker work pool - Single container: Dask LocalCluster works out of the box; optionally set CPU limits via

container_create_kwargs

and size your cluster accordingly. - External cluster: ensure the container can reach the scheduler (network/DNS/ports). If you’re on Kubernetes, consider dask_kubernetes and point to that scheduler. Docs - Prefect + Dask integration: https://docs.prefect.io/integrations/prefect-dask - Dask LocalCluster: https://docs.dask.org/en/stable/deploying-python.html#localcluster If you share your workload characteristics (CPU vs IO, data sizes, expected parallelism, infra), I can recommend the best runner and a concrete config.

Great use case. In Prefect 3, a flow’s task runner is fixed for that flow, so if Dask is only beneficial for a later stage, the cleanest pattern is to put that stage into a subflow that uses Dask, and call it from your main flow. Option 1: Subflow that uses Dask (recommended) - Keep your top-level flow on the default runner (or ProcessPool/ThreadPool). - Define a subflow with DaskTaskRunner and call it when you reach the heavy fan-out stage.

# pip install "prefect-dask" "dask[distributed]"

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

@task
def transform(x): 
    return x * x

# Use an external Dask cluster
@flow(task_runner=DaskTaskRunner(address="<tcp://dask-scheduler:8786>"))
def heavy_stage(keys):
    futures = [transform.submit(k) for k in keys]
    return [f.result() for f in futures]

@flow
def main_flow():
    # upstream work on a simple runner
    keys = list(range(1000))
    # switch to Dask only here
    results = heavy_stage(keys)
    return results

Notes: - External cluster: ensure the Docker flow container can reach the scheduler host/port (8786). Configure Docker network_mode or routes as needed. - LocalCluster instead: if you just need multi-process parallelism in one container, use LocalCluster:

from dask.distributed import LocalCluster
  @flow(task_runner=DaskTaskRunner(cluster_class=LocalCluster,
                                   cluster_kwargs={"n_workers": 4, "threads_per_worker": 1, "processes": True}))
  def heavy_stage(keys): ...

- Install dependencies in your image:

prefect-dask

and

dask[distributed]

. Option 2: Split into two deployments - Make two flows/deployments: upstream_flow and heavy_stage_flow. - Upstream finishes, then triggers the downstream deployment (which uses DaskTaskRunner) via an Automation (on Completed event → Run Deployment) or from code. - Benefits: isolate environments (e.g., smaller image upstream, Dask-enabled image downstream) and run the heavy step on a different work pool/infra. Example trigger from code:

from prefect import flow
from prefect.deployments import run_deployment

@flow
def upstream_flow():
    # do stuff...
    run_deployment(name="heavy-stage-flow/heavy-stage-deployment", parameters={"keys": list(range(1000))})

Practical tips - Don’t pass huge objects between parent and subflow; store them (S3/GCS/local) and pass references/keys. - For Docker workers with external Dask: - Ensure the container can reach the scheduler (network, DNS, firewall). - If you use a VPN or sidecar network, set

network_mode

appropriately in your work pool job template. - For CPU-bound work on a single machine without Dask, a simpler alternative is ProcessPoolTaskRunner in the main flow or subflow. Docs - Prefect + Dask integration: https://docs.prefect.io/integrations/prefect-dask If you share your infra (single host vs cluster, external Dask available or not) and rough workload size, I can suggest exact config for your work pool and a minimal example tailored to it.