Joe D
09/09/2025, 5:31 PMMarvin
09/09/2025, 5:31 PMMarvin
09/09/2025, 5:34 PMfrom prefect import task
@task(persist_result=True)
def compute(...):
...
- At the flow level (to persist the flow’s return value):
from prefect import flow
@flow(persist_result=True)
def my_flow(...):
...
- Use a shared result storage when runs span multiple workers/containers:
from prefect import flow, task
from prefect.filesystems import RemoteFileSystem
result_store = RemoteFileSystem(basepath="<s3://my-bucket/prefect-results/>") # or GCS/Azure/NFS
@task(persist_result=True, result_storage=result_store)
def t(x):
return x + 1
@flow(persist_result=True, result_storage=result_store)
def f():
return t(1)
Make sure the workers have credentials and network access to that store.
- If you frequently need persisted results, consider setting a profile/setting so results persist by default, or set persist_result=True on the decorators you use most. See docs for details.
When you do NOT need persistence
- Within a single flow run on the same worker with the default task runner, upstream results are passed in‑memory and do not need persistence. If you still see this error in that scenario, it usually indicates the process crashed/restarted or you’re crossing a process/infra boundary (e.g., subflows on different workers).
Docs
- Results, persistence, and storage: Prefect 3.x: Results
To help pinpoint your case, could you share:
- Where this error shows up (worker logs, downstream task, external client calling `state.result()`/`.get()`)?
- Your Prefect version, flow/task decorators (any `persist_result`/`result_storage` set?), and your worker/work pool/infrastructure.
- Whether you’re using subflows, retries, or a non-default task runner (e.g., ProcessPoolTaskRunner).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