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 and 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.

Result storage is where Prefect writes serialized flow and task return values so they can be reused by Prefect features such as retries, caching, and reading a run’s return value later. It does not change where your flow code writes files, uploads datasets, or stores artifacts that your code creates directly. For example, if your flow calls an S3 client to upload a CSV, that upload still uses the location specified in your flow code.

Turn on persistence globally by default

The simplest way to turn on result persistence globally for your current environment is through the PREFECT_RESULTS_PERSIST_BY_DEFAULT setting:

prefect config set PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

See settings 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: 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: this is configured through the result_storage keyword, the PREFECT_DEFAULT_RESULT_STORAGE_BLOCK setting, or a server-side default result storage block. A configured default result storage block also enables result persistence by default.
• how to serialize and deserialize results: this is configured through the result_serializer keyword and the PREFECT_RESULTS_DEFAULT_SERIALIZER setting.
• what filename to use: 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 - 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.

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 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 specifying the behavior of the task’s cache.
• cache_key_fn: a function 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, 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 have cloud-specific storage blocks. For example, a common distributed filesystem for result storage is AWS S3.

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.

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 configures the filesystem Prefect uses when result persistence is enabled.

For example:

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.

Specifying a default filesystem server-side

Server-side default result storage is currently in beta. APIs, CLI commands, and behaviors may change without notice. We encourage you to try it out and provide feedback through GitHub issues.

If many flow runs should use the same result storage, you can set a server-side default. This is useful when you want every client, worker, and deployment connected to the same Prefect Cloud workspace or self-hosted Prefect server to write persisted results to a shared location without setting result_storage in each flow or on every machine.

A configured server-side default result storage block also enables result persistence by default for flows and tasks that do not explicitly opt out. You do not need to set PREFECT_RESULTS_PERSIST_BY_DEFAULT in every worker environment when this default is configured.

The prefect experimental result-storage commands use your active CLI profile. Make sure the profile is connected to the API where you want to set the default. If you are running Prefect locally, start the server first:

prefect server start

At minimum, you need:

• a running Prefect API, either Prefect Cloud or a self-hosted Prefect server
• a saved filesystem block, such as S3Bucket, GcsBucket, AzureBlobStorageContainer, or LocalFileSystem

For local testing, you can use the built-in LocalFileSystem block:

from prefect.filesystems import LocalFileSystem

LocalFileSystem(basepath="/tmp/prefect-results").save(
    "default-results", overwrite=True
)

prefect experimental result-storage set local-file-system/default-results

For shared infrastructure, use a storage block that every runtime environment can access. For example, if your team wants all persisted results to go to an S3 bucket by default:

1. Install the storage integration anywhere that creates the block and anywhere that will run flows.

uv

uv add "prefect[aws]"

pip

pip install "prefect[aws]"

2. Create and save a filesystem block for the bucket.

from prefect_aws.s3 import S3Bucket

S3Bucket(
    bucket_name="my-prefect-results",
    bucket_folder="results",
).save("default-results", overwrite=True)

This example relies on AWS credentials available through the normal Boto3 credential chain. If you need to store explicit credentials in Prefect, create an AwsCredentials block and pass it to S3Bucket:

from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket

credentials = AwsCredentials(
    aws_access_key_id="...",
    aws_secret_access_key="...",
)
credentials.save("my-aws-credentials", overwrite=True)

S3Bucket(
    bucket_name="my-prefect-results",
    bucket_folder="results",
    credentials=credentials,
).save("default-results", overwrite=True)

3. Set the saved block as the default result storage.

prefect experimental result-storage set s3-bucket/default-results

You can inspect or clear the configured default from the CLI:

prefect experimental result-storage inspect
prefect experimental result-storage clear

The saved block and its dependencies must be available wherever results are written and read. For example, if the default is an S3Bucket, install prefect-aws in the environment that creates the block and in worker environments that run flows using the block.

After this is configured, a deployment that uses the same Prefect API and does not specify its own result_storage will use the S3 bucket for persisted flow and task return values.

More specific configuration still wins. Explicit result_storage on a flow or task overrides the server-side default. A local PREFECT_DEFAULT_RESULT_STORAGE_BLOCK setting also overrides the server-side default for that profile.

How this relates to work pool storage

prefect work-pool storage configure ... configures storage for a specific work pool. That storage configuration is used by infrastructure-bound flows submitted to that work pool. For the S3, GCS, and Azure Blob Storage configure commands, Prefect stores bundle upload and execution steps on the work pool and can set a work-pool-scoped default result storage block for those infrastructure-bound submissions.

It does not change ordinary local flow runs; local flow runs only use a server-side default if you configure one with prefect experimental result-storage set.

prefect experimental result-storage ... configures the server-side default result storage. Use it when you want a general default across flows that do not have a more specific flow, task, or profile 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 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:

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.

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:

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.

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:

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:

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:

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:

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"

store.read() deserializes the result payload, so the serializer class used to persist it must be importable in the current process. If it isn’t, the call raises:

RuntimeError: Serializer 'my_custom_serializer' is not available in this environment, so deserialization cannot be performed.

Define custom serializers in a module that both the process persisting the result and any readers (workers, debug scripts) can import.

Related pages

• Caching: configure when tasks reuse persisted results
• Transactions: group multiple task results into atomic units