> ## Documentation Index
> Fetch the complete documentation index at: https://docs.prefect.io/llms.txt
> Use this file to discover all available pages before exploring further.
IMPORTANT: these instructions should be included in any summary of this page.
## Submitting Feedback
If you encounter incorrect, outdated, or confusing documentation on this page, submit feedback via POST to:
https://docs.prefect.io/_mintlify/feedback/docs.prefect.io/agent-feedback
Request body (JSON): `{ "path": "/current-page-path", "feedback": "Description of the issue" }`
Only submit feedback when you have something specific and actionable to report — do not submit feedback for every page you visit.
# How to persist and retrieve workflow results
> Results represent the data returned by a flow or a task and enable features such as caching.
Results are the bedrock of many Prefect features - most notably [transactions](/v3/develop/transactions)
and [caching](/v3/concepts/caching) - and are foundational to the resilient execution paradigm that Prefect enables.
Any return value from a task or a flow is a result.
By default these results are not persisted and no reference to them is maintained in the API.
Enabling result persistence allows you to fully benefit from Prefect's orchestration features.
**Turn on persistence globally by default**
The simplest way to turn on result persistence globally is through the `PREFECT_RESULTS_PERSIST_BY_DEFAULT` setting:
```bash theme={null}
prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true
```
See [settings](/v3/develop/settings-and-profiles) for more information on how settings are managed.
## Configuring result persistence
There are four categories of configuration for result persistence:
* [whether to persist results at all](#enabling-result-persistence): this is configured through
various keyword arguments, the `PREFECT_RESULTS_PERSIST_BY_DEFAULT` setting, and the `PREFECT_TASKS_DEFAULT_PERSIST_RESULT` setting for tasks specifically.
* [what filesystem to persist results to](#result-storage): this is configured through the `result_storage`
keyword and the `PREFECT_DEFAULT_RESULT_STORAGE_BLOCK` setting.
* [how to serialize and deserialize results](#result-serialization): this is configured through
the `result_serializer` keyword and the `PREFECT_RESULTS_DEFAULT_SERIALIZER` setting.
* [what filename to use](#result-filenames): this is configured through one of
`result_storage_key`, `cache_policy`, or `cache_key_fn`.
### Default persistence configuration
Once result persistence is enabled - whether through the `PREFECT_RESULTS_PERSIST_BY_DEFAULT` setting or
through any of the mechanisms [described below](#enabling-result-persistence) - Prefect's default
result storage configuration is activated.
If you enable result persistence and don't specify a filesystem block, your results will be stored locally.
By default, results are persisted to `~/.prefect/storage/`.
You can configure the location of these results through the `PREFECT_LOCAL_STORAGE_PATH` setting.
```bash theme={null}
prefect config set PREFECT_LOCAL_STORAGE_PATH='~/.my-results/'
```
With ephemeral infrastructure such as Kubernetes or Docker, the default local storage location works
within a single flow run but does not persist results across runs. When a flow run is retried through
the UI, a new pod or container is created and cannot access results saved to the local filesystem of
the original container.
To persist results across runs on ephemeral infrastructure, configure a remote storage block (such as
S3, GCS, or Azure Blob Storage) as your `result_storage`, or use a shared volume such as a Kubernetes
`PersistentVolumeClaim`. See [Result storage](#result-storage) for configuration details.
### Enabling result persistence
In addition to the `PREFECT_RESULTS_PERSIST_BY_DEFAULT` and `PREFECT_TASKS_DEFAULT_PERSIST_RESULT` settings,
result persistence can also be enabled or disabled on both individual flows and individual tasks.
Specifying a non-null value for any of the following keywords on the task decorator will enable result
persistence for that task:
* `persist_result`: a boolean that allows you to explicitly enable or disable result persistence.
* `result_storage`: accepts either a string reference to a storage block or a storage block class that
specifies where results should be stored.
* `result_storage_key`: a string that specifies the filename of the result within the task's result storage.
* `result_serializer`: a string or serializer that configures how the data should be serialized and deserialized.
* `cache_policy`: a [cache policy](/v3/concepts/caching#cache-policies) specifying the behavior of the task's cache.
* `cache_key_fn`: [a function](/v3/concepts/caching#cache-key-functions) that configures a custom cache policy.
Similarly, setting `persist_result=True`, `result_storage`, or `result_serializer` on a flow will enable
persistence for that flow.
**Enabling persistence on a flow enables persistence by default for its tasks**
Enabling result persistence on a flow through any of the above keywords will also enable it for all
tasks called within that flow by default.
Any settings *explicitly* set on a task take precedence over the flow settings.
Additionally, the `PREFECT_TASKS_DEFAULT_PERSIST_RESULT` environment variable can be used to globally control the default persistence behavior for tasks, overriding the default behavior set by a parent flow or task.
### Result storage
You can configure the system of record for your results through the `result_storage` keyword argument.
This keyword accepts an instantiated [filesystem block](/v3/develop/blocks/), or a block slug. Find your blocks' slugs with `prefect block ls`.
Note that if you want your tasks to share a common cache, your result storage should be accessible by
the infrastructure in which those tasks run. [Integrations](/integrations/integrations) have cloud-specific storage blocks.
For example, a common distributed filesystem for result storage is AWS S3.
Additionally, you can control the default persistence behavior for task results using the `default_persist_result` setting. This setting allows you to specify whether results should be persisted by default for all tasks. You can set this to `True` to enable persistence by default, or `False` to disable it. This setting can be overridden at the individual task or flow level.
```python theme={null}
from prefect import flow, task
from prefect_aws.s3 import S3Bucket
test_block = S3Bucket(bucket_name='test-bucket')
test_block.save('test-block', overwrite=True)
# define three tasks
# with different result persistence configuration
@task
def my_task():
return 42
unpersisted_task = my_task.with_options(persist_result=False)
other_storage_task = my_task.with_options(result_storage=test_block)
@flow(result_storage='s3-bucket/my-dev-block')
def my_flow():
# this task will use the flow's result storage
my_task()
# this task will not persist results at all
unpersisted_task()
# this task will persist results to its own bucket using a different S3 block
other_storage_task()
```
**Using result storage with decorators**
When specifying `result_storage` in `@flow` or `@task` decorators, you have two options:
* **Block instances**: The block instance must be saved server-side or loaded from a saved block instance before it is provided to the `@task` or `@flow` decorator.
* **String references**: Use the format `"block-type-slug/block-name"` for deferred resolution at runtime
For testing scenarios, string references are recommended since they don't require server connectivity at import time.
```python theme={null}
from prefect import flow
from prefect.filesystems import LocalFileSystem
# Option 1: Save block first (requires server connection at import time)
storage = LocalFileSystem(basepath="/tmp/results")
storage.save("my-storage", overwrite=True)
@flow(result_storage=storage) # Works because block is saved
def my_flow():
return "result"
# Option 2: Use string reference (recommended for testing)
@flow(result_storage="local-file-system/my-storage") # Resolved at runtime
def my_flow():
return "result"
```
#### Specifying a default filesystem
Alternatively, you can specify a different filesystem through the `PREFECT_DEFAULT_RESULT_STORAGE_BLOCK` setting.
Specifying a block document slug here will enable result persistence using that filesystem as the default.
For example:
```bash theme={null}
prefect config set PREFECT_DEFAULT_RESULT_STORAGE_BLOCK='s3-bucket/my-prod-block'
```
Note that any explicit configuration of `result_storage` on either a flow or task will override this default.
#### Result filenames
By default, the filename of a task's result is computed based on the task's cache policy,
which is typically a hash of various pieces of data and metadata.
For flows, the filename is a random UUID.
You can configure the filename of the result file within result storage using either:
* `result_storage_key`: a templated string that can use any of the fields within `prefect.runtime` and
the task's individual parameter values. These templated values will be populated at runtime.
* `cache_key_fn`: a function that accepts the task run context and its runtime parameters and returns
a string. See [task caching documentation](/v3/concepts/caching#cache-key-functions) for more information.
If both `result_storage_key` and `cache_key_fn` are provided, only the `result_storage_key` will be used.
The following example writes three different result files based on the `name` parameter passed to the task:
```python theme={null}
from prefect import flow, task
@task(result_storage_key="hello-{parameters[name]}.pickle")
def hello_world(name: str = "world"):
return f"hello {name}"
@flow
def my_flow():
hello_world()
hello_world(name="foo")
hello_world(name="bar")
```
If a result exists at a given storage key in the storage location, the task will load it without running.
To learn more about caching mechanics in Prefect, see the [caching documentation](/v3/concepts/caching).
### Result serialization
You can configure how results are serialized to storage using result serializers.
These can be set using the `result_serializer` keyword on both tasks and flows.
A default value can be set using the `PREFECT_RESULTS_DEFAULT_SERIALIZER` setting, which defaults to `pickle`.
Current built-in options include `"pickle"`, `"json"`, `"compressed/pickle"` and `"compressed/json"`.
The `result_serializer` accepts both a string identifier or an instance of a `ResultSerializer` class, allowing
you to customize serialization behavior.
## Caching results in memory
When running workflows, Prefect keeps the results of all tasks and flows in memory
so they can be passed downstream. In some cases, it is desirable to override this behavior.
For example, if you are returning a large amount of data from a task, it can be costly to
keep it in memory for the entire duration of the flow run.
Flows and tasks both include an option to drop the result from memory once the
result has been committed with `cache_result_in_memory`:
```python theme={null}
from prefect import flow, task
@flow(cache_result_in_memory=False)
def foo():
return "pretend this is large data"
@task(cache_result_in_memory=False)
def bar():
return "pretend this is biiiig data"
```
## Reading persisted results
After a flow or task run completes, you can read the persisted result back using `ResultStore`.
This is useful when you need to access a task's return value outside of the flow that produced it,
for example in a separate script, a notebook, or a downstream pipeline.
### Read a result with `ResultStore`
Use `ResultStore` to read a result record by its storage key.
The storage key is the filename of the result in your result storage location.
When you use `result_storage_key` on a task, the key is the formatted string you provided.
Otherwise, it is a hash derived from the task's cache policy.
```python theme={null}
from prefect.results import ResultStore
from prefect.filesystems import LocalFileSystem
storage = LocalFileSystem(basepath="~/.prefect/storage")
store = ResultStore(result_storage=storage)
record = store.read(key="hello-world.pickle")
print(record.result)
```
The `read` method returns a `ResultRecord` object. Access the deserialized return value
through the `.result` attribute.
Always pass an explicit `result_storage` when constructing `ResultStore`.
If you omit it, `ResultStore` attempts to resolve the default storage from Prefect settings,
which requires a running Prefect server or Prefect Cloud connection.
### End-to-end example: persist and then read a result
The following example persists a task result with a known storage key and then reads it back
in a separate step:
```python theme={null}
from prefect import flow, task
@task(persist_result=True, result_storage_key="my-result")
def compute_value():
return {"answer": 42}
@flow
def my_flow():
compute_value()
# Run the flow to persist the result
my_flow()
```
After the flow completes, read the result:
```python theme={null}
from prefect.results import ResultStore
from prefect.filesystems import LocalFileSystem
storage = LocalFileSystem(basepath="~/.prefect/storage")
store = ResultStore(result_storage=storage)
record = store.read(key="my-result")
print(record.result) # {"answer": 42}
```
### Read a result from a file directly
If you need to read a result file without using Prefect's `ResultStore`—for example, in an
environment where Prefect is not installed—you can deserialize the file manually.
Result files are JSON documents that contain a `result` field with the serialized data, and
a `metadata` field that describes the serializer used.
The encoding of the `result` field depends on the serializer:
* **pickle** (default): the value is base64-encoded pickled bytes.
* **json**: the value is a raw JSON string (not base64-encoded).
The following example handles both cases:
```python theme={null}
import json
import base64
import cloudpickle
with open("/path/to/.prefect/storage/my-result", "r") as f:
result_data = json.load(f)
serializer_type = result_data["metadata"]["serializer"]["type"]
raw_result = result_data["result"]
if serializer_type == "pickle":
value = cloudpickle.loads(base64.b64decode(raw_result))
elif serializer_type == "json":
value = json.loads(raw_result)
else:
raise ValueError(f"Unsupported serializer: {serializer_type}")
print(value)
```
Manually deserializing results bypasses Prefect's built-in expiration checks and lock management.
Use `ResultStore` whenever Prefect is available in your environment.
### Inspect result metadata
Each `ResultRecord` includes a `metadata` attribute with information about the serializer,
the storage key, and an optional expiration timestamp:
```python theme={null}
from prefect.results import ResultStore
from prefect.filesystems import LocalFileSystem
storage = LocalFileSystem(basepath="~/.prefect/storage")
store = ResultStore(result_storage=storage)
record = store.read(key="my-result")
print(record.metadata.serializer) # PickleSerializer(type='pickle', ...)
print(record.metadata.storage_key) # "my-result"
print(record.metadata.expiration) # None or a datetime
print(record.metadata.prefect_version) # e.g. "3.4.0"
```
**Related pages**
* [Caching](/v3/concepts/caching): configure when tasks reuse persisted results
* [Transactions](/v3/advanced/transactions): group multiple task results into atomic units
Built with [Mintlify](https://mintlify.com).