Olve Heitmann
09/14/2025, 4:05 PMMarvin
09/14/2025, 4:05 PMMarvin
09/14/2025, 4:08 PMpersist_result=True and point result_storage to shared storage (e.g., S3, GCS, Azure blob, NFS).
- Persist the cache key/index: provide a cache_policy with key_storage set to a shared filesystem path (e.g., S3/GCS/Azure/NFS). This is what lets any worker find the existing result by cache key.
- Optional (recommended for distributed workers): set a stricter isolation level and a distributed lock to prevent duplicate work.
- isolation_level="SERIALIZABLE" with a LockManager that all workers can reach (e.g., filesystem-based on shared NFS, or a Redis-based lock manager via the prefect-redis integration).
Example
from datetime import timedelta
from prefect import task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import RemoteFileSystem
# If you want distributed locking, use a LockManager reachable by all workers
shared_cache_policy = CachePolicy(
key_storage=RemoteFileSystem(basepath="<s3://my-bucket/prefect/cache>"),
# isolation_level="SERIALIZABLE",
# lock_manager=... # e.g., a Redis-based lock manager
)
shared_results = RemoteFileSystem(basepath="<s3://my-bucket/prefect/results>")
@task(
persist_result=True,
result_storage=shared_results,
cache_policy=shared_cache_policy,
cache_expiration=timedelta(days=1),
)
def expensive_task(x):
...
Key points
- Worker pools are just a scheduling boundary; cache reuse depends on shared storage and cache keys, not the pool itself.
- The cache key must match. If your cache policy includes the task source, input values, etc., changing code or inputs will produce a different key and skip reuse.
- The default cache is an in-process LRU cache, so it’s not reusable across processes, workers, or pools.
Docs
- Caching in Prefect 3.x: Caching (Prefect 3.x)
- Redis integration (for distributed locks/key stores): prefect-redis integration
If you share your environment (e.g., S3 vs GCS vs NFS) I can tailor the exact setup for your workers and work pools.Olve Heitmann
09/14/2025, 4:15 PMOlve Heitmann
09/14/2025, 4:30 PMMarvin
09/14/2025, 4:30 PMMarvin
09/14/2025, 4:35 PMfrom datetime import timedelta
from prefect import task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import RemoteFileSystem
# For distributed locking, see the prefect-redis integration docs
shared_results = RemoteFileSystem(basepath="<s3://my-bucket/prefect/results>")
shared_cache = RemoteFileSystem(basepath="<s3://my-bucket/prefect/cache>")
@task(
persist_result=True,
result_storage=shared_results,
cache_policy=CachePolicy(
key_storage=shared_cache,
# isolation_level="SERIALIZABLE",
# lock_manager=... # e.g., a Redis-based lock manager
),
cache_expiration=timedelta(days=1),
)
def expensive_task(x):
...
- Ensure your Docker image/containers have credentials to access the storage (e.g., AWS env vars).
- If you want to avoid duplicate “first writer wins” races when the same cache key hits both pools at once, set isolation_level="SERIALIZABLE" and supply a distributed lock_manager (e.g., Redis). See prefect-redis integration.
Option B: use a shared Docker volume (NFS/EFS/host bind) mounted into every flow container
- Pros: simple if all containers run on the same host or a shared NFS.
- How:
- Mount the same host path into every flow container in both pools at the same container path, e.g., /mnt/prefect
- Point result storage and cache key storage at directories inside that mount
- Use filesystem locks to prevent duplicate work
Example
from pathlib import Path
from datetime import timedelta
from prefect import task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import LocalFileSystem
from prefect.locking.filesystem import FileSystemLockManager
shared_results_fs = LocalFileSystem(basepath="/mnt/prefect/results")
shared_cache_fs = LocalFileSystem(basepath="/mnt/prefect/cache")
lock_manager = FileSystemLockManager(lock_files_directory=Path("/mnt/prefect/locks"))
@task(
persist_result=True,
result_storage=shared_results_fs,
cache_policy=CachePolicy(
key_storage=shared_cache_fs,
isolation_level="SERIALIZABLE",
lock_manager=lock_manager,
),
cache_expiration=timedelta(days=1),
)
def expensive_task(x):
...
Mounting the volume into flow containers
- Prefect’s Docker work pool uses a base job template that defines container settings (image, env, volumes, etc.). You can:
1) Inspect the current template:
prefect work-pool inspect <your-docker-pool>
2) Save and edit the template JSON to include a “volumes” variable and wire it to the Docker run configuration if it isn’t there already.
3) Update the pool:
prefect work-pool create "<your-docker-pool>" --type docker --base-job-template ./docker-template.json --overwrite
4) Set the volumes via job variables on your deployments (or pool defaults). Typical format is Docker-style bind mounts (e.g., "/srv/prefect:/mnt/prefect:rw"), but confirm the exact field name/shape in your pool’s template schema shown by inspect.
Notes and gotchas
- Worker pools are just scheduling boundaries; cache reuse depends on shared storage and consistent cache keys.
- The default cache is in-memory per process — not shared — so you must configure storage as shown above.
- If code or inputs change, the cache key will change; only equal keys reuse results. Customize cache_key_fn if you need tighter control.Marvin
09/14/2025, 4:35 PMprefect work-pool inspect <pool>
…I can give you exact @task settings and a precise base job template/job_variables snippet for your setup.Olve Heitmann
09/14/2025, 9:12 PMMarvin
09/14/2025, 9:13 PMMarvin
09/14/2025, 9:14 PMcache_policy.key_storage to your shared volume
- persist_result=True and result_storage to your shared volume
Do I need multiple filesystems?
- Not strictly. You can:
- Use one LocalFileSystem for both key storage and results (they’ll mix in one directory), or
- Use two LocalFileSystem instances pointing at different subdirectories under the same mount. This is cleaner and lets you clear cache without touching results.
- Functionally, both work. Operationally, separate subdirs is nicer.
Is /mnt required?
- No. It’s just a common Linux convention. Use any consistent container path you’ve mounted, e.g. /data/prefect or /srv/prefect. The key is: every flow container in both pools should mount the same host path to the same container path.
Minimal, clean setup on a single shared mount
- Assume you mounted the host path to /srv/prefect inside each flow container.
from datetime import timedelta
from pathlib import Path
from prefect import task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import LocalFileSystem
from prefect.locking.filesystem import FileSystemLockManager
# Separate subdirs under the same shared mount
results_fs = LocalFileSystem(basepath="/srv/prefect/results")
cache_fs = LocalFileSystem(basepath="/srv/prefect/cache")
locks_dir = Path("/srv/prefect/locks")
@task(
persist_result=True, # store results so other containers can read them
result_storage=results_fs, # write results to shared volume
cache_policy=CachePolicy(
key_storage=cache_fs, # write cache index to shared volume
isolation_level="SERIALIZABLE", # optional but recommended to avoid duplicate work
lock_manager=FileSystemLockManager(lock_files_directory=locks_dir),
),
cache_expiration=timedelta(days=1),
)
def expensive_task(x):
# your work here
return ...
Notes and tips
- You can point both result_storage and key_storage at the same LocalFileSystem (same basepath) if you want to keep it ultra-simple; I still recommend separate subfolders to avoid clutter and simplify maintenance.
- The lock manager + SERIALIZABLE helps when both pools receive the same cacheable task at the same time — only one will compute; the other will wait and reuse.
- cache_result_in_memory=True (default) only helps within a single process; it doesn’t replace persisted storage.
- Make sure your pool’s base job template or job variables mount the shared host directory into the container at the exact path you use above; you’ve said that’s already done.
Docs for reference
- Caching in Prefect 3.x: Caching (Prefect 3.x)
- Redis option (not required in your single-host setup, but useful for distributed locks): prefect-redis integration
If you share the container path you’ve mounted (e.g., /srv/prefect vs something else), I can tweak the snippet exactly to your tree.Olve Heitmann
09/14/2025, 10:01 PMMarvin
09/14/2025, 10:01 PMMarvin
09/14/2025, 10:02 PMfrom datetime import timedelta
from pathlib import Path
from prefect import flow, task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import LocalFileSystem
from prefect.locking.filesystem import FileSystemLockManager
results_fs = LocalFileSystem(basepath="/srv/prefect/results")
cache_fs = LocalFileSystem(basepath="/srv/prefect/cache")
locks_dir = Path("/srv/prefect/locks")
@task(
persist_result=True,
result_storage=results_fs,
cache_policy=CachePolicy(
key_storage=cache_fs,
isolation_level="SERIALIZABLE",
lock_manager=FileSystemLockManager(lock_files_directory=locks_dir),
),
cache_expiration=timedelta(days=1),
)
def expensive_task(x):
# heavy work
return f"result for {x}"
# A “refresh” view of the same task
refresh_expensive_task = expensive_task.with_options(refresh_cache=True)
@flow
def my_flow(items):
# normal cached calls
results = [expensive_task.submit(i) for i in items]
# ... do the rest of your flow
# At the end, refresh selectively for next runs
to_refresh = [items[0]] # pick whatever keys you want to refresh
for i in to_refresh:
# submit refresh; this recomputes and updates the persisted cache
refresh_expensive_task.submit(i)
Notes
- You can refresh multiple keys by looping and calling the refresh variant with each input.
- If you don’t need an immediate refresh, you can rely on cache_expiration so keys expire naturally.
- If you rarely need to refresh, you can place the refresh logic in a separate “maintenance” flow that only runs when you want to force updates.
Less recommended alternatives
- Manually deleting cache files from the shared volume is possible but brittle; the file layout is an internal detail and easy to get wrong.
- Permanently setting refresh_cache=True on the decorator forces recompute every time (usually not what you want).
Docs
- Caching in Prefect 3.x: Caching (Prefect 3.x)
If you share how you choose which items to refresh (e.g., all, a filtered subset, time-based), I can sketch the exact refresh loop or a small maintenance flow for your case.Olve Heitmann
09/14/2025, 10:15 PMMarvin
09/14/2025, 10:16 PMMarvin
09/14/2025, 10:17 PMrefresh_cache=True behaves for a task in Prefect 3.x:
Short answers to your 1/2/3:
1) Does it empty the current cache? No. It bypasses the cache read, recomputes the task, and then overwrites/updates the persisted cache entry for that key. It does not “delete” the cache first; it replaces it.
2) Does it just refresh the expiry on the same data? Not as a “touch.” The expiration gets reset because a new record is written after recomputation. There’s no “extend TTL without recompute” operation.
3) Does it recompute and persist but not use the result in this run? By default, no — the call returns the newly computed result like any task call. If you don’t want to use it in the current flow, call the refresh at the end (or in a maintenance flow) and ignore the returned future so it doesn’t affect earlier downstream logic.
A practical “refresh at end” pattern
- Use the cached task as usual early in the flow.
- After your downstream work is done, call a “refresh” view of the same task to rebuild the cache for next time, and discard the result.
from datetime import timedelta
from pathlib import Path
from prefect import flow, task
from prefect.cache_policies import CachePolicy
from prefect.filesystems import LocalFileSystem
from prefect.locking.filesystem import FileSystemLockManager
results_fs = LocalFileSystem(basepath="/srv/prefect/results")
cache_fs = LocalFileSystem(basepath="/srv/prefect/cache")
locks_dir = Path("/srv/prefect/locks")
@task(
persist_result=True,
result_storage=results_fs,
cache_policy=CachePolicy(
key_storage=cache_fs,
isolation_level="SERIALIZABLE", # optional but helps avoid duplicate work
lock_manager=FileSystemLockManager(lock_files_directory=locks_dir),
),
cache_expiration=timedelta(days=1),
)
def expensive_task(x):
# heavy work
return f"result for {x}"
# A “refresh” view that bypasses read and recomputes
refresh_expensive_task = expensive_task.with_options(refresh_cache=True)
@flow
def my_flow(items):
# Use the cached task for normal work
results = [expensive_task.submit(i) for i in items]
# ... downstream work using `results` ...
# At the end, rebuild cache for a subset or all items for next runs
for i in items[:2]: # choose which keys to refresh
_ = refresh_expensive_task.submit(i) # recompute + update cache; ignore output
Notes
- If concurrent runs might hit the same key while you refresh, using isolation_level="SERIALIZABLE" with FileSystemLockManager ensures one writer and consistent reads.
- There isn’t a first-class “invalidate this one cache entry without recompute” API. Manual deletion of specific cache files is possible but brittle; recomputing with refresh_cache=True is the safer route.
- If you need to force new keys (e.g., schema change) you can add a salt/version into your cache_key_fn so old and new entries don’t collide.
Docs for reference
- Caching in Prefect 3.x: Caching (Prefect 3.x)
If you want, I can show how to target a specific cache key (via a custom cache_key_fn) so you can refresh exactly the entries you want.Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.
Powered by