Short answer: use Prefect’s compressed pickle serializer, or better yet store the DataFrame in a columnar format (Parquet/IPC) and cache only a reference. Details and recommendations (Prefect 3.x): - Best built-in serializer for large pandas/polars DataFrames: CompressedPickleSerializer (

'compressed/pickle'

). It uses cloudpickle + compression and handles complex Python objects reliably. - Avoid JSON serializers for DataFrames; they only support JSON-compatible types and will either fail or bloat. - For very large data, the recommended pattern is to write the DataFrame to external storage (e.g., Parquet in S3/GCS/local) and have the task return the URI. Prefect then caches/persists that small string cheaply and you re-load as needed. Practical tips - Set

cache_result_in_memory=False

to avoid holding huge results in memory in the worker after persistence. - Configure

persist_result=True

and a result storage target if you want cross-run caching/persistence. - You can inherit these settings from the flow or set them per-task. Example A: Persist the actual DataFrame (simple and general, but heavier)

from prefect import task, flow

@task(
    persist_result=True,
    result_serializer="compressed/pickle",
    cache_result_in_memory=False,
)
def build_df():
    import pandas as pd
    df = pd.DataFrame({"a": range(1_000_000)})
    return df

@flow
def main():
    df = build_df()
    return df

if __name__ == "__main__":
    main()

Example B (recommended for very large DataFrames): Write Parquet and cache the path

from prefect import task, flow
from pathlib import Path

@task(
    persist_result=True,
    # returning a path/URI is small, so json or pickle are both fine
    result_serializer="json",
    cache_result_in_memory=False,
)
def write_df_to_parquet() -> str:
    import pandas as pd
    df = pd.DataFrame({"a": range(10_000_000)})
    out = Path("data/df.parquet")
    out.parent.mkdir(parents=True, exist_ok=True)
    df.to_parquet(out)  # for polars: df.write_parquet(out)
    return str(out)

@task
def read_parquet(path: str):
    import pandas as pd
    return pd.read_parquet(path)

@flow
def main():
    path = write_df_to_parquet()
    df = read_parquet(path)
    return df

if __name__ == "__main__":
    main()

Notes - If you’re persisting actual results, consider remote result storage (e.g., S3/GCS/Azure) so the cache is shared across runs/workers. - You can globally default to compressed pickle with

PREFECT_RESULTS_DEFAULT_SERIALIZER=compressed/pickle

. Docs for reference - Results (persistence, storage, serializers): Prefect 3.x Results - Serializers API: Serializers API - Caching concepts: Caching Follow-up question: About how large are your DataFrames and where do you want to store persisted results (local disk, S3, GCS, etc.)? I can tailor an example with the right storage block and file format.

Great question. In Prefect 3.x, cache expiration and file expiration are separate: -

cache_expiration

only controls whether Prefect will reuse a cached result; it does not delete the underlying file. - Prefect does not automatically delete persisted result files (e.g., Parquet) for you. Recommended ways to expire/clean up Parquet files 1) Use your storage’s lifecycle policies (best/low-touch) - S3: set a lifecycle rule to expire/delete objects under a given prefix after N days. - GCS/Azure: similar lifecycle/retention policies can auto-delete by prefix/age. - Pattern: write your Parquet to a time-based prefix (e.g.,

.../parquet/YYYY/MM/DD/...

) so your lifecycle rule can target that prefix safely. 2) Schedule a Prefect “janitor” flow to delete old files - Run it daily/weekly to remove files older than your retention period. - Example (S3 via fsspec/s3fs):

from datetime import datetime, timedelta, timezone
from prefect import flow, task

@task
def delete_old_s3(prefix: str, days: int, bucket: str = "my-bucket"):
    import fsspec
    fs = fsspec.filesystem("s3")  # requires s3fs
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    # list files under prefix
    files = fs.find(f"s3://{bucket}/{prefix}")
    to_delete = []
    for path in files:
        info = <http://fs.info|fs.info>(path)  # includes LastModified
        last_modified = info.get("LastModified") or info.get("modified")  # provider dependent
        if last_modified and last_modified < cutoff:
            to_delete.append(path)

    for p in to_delete:
        fs.rm(p)
    return len(to_delete)

@flow
def cleanup_s3(prefix: str, days: int):
    deleted = delete_old_s3(prefix, days)
    print(f"Deleted {deleted} old files")

- Example (local filesystem):

from datetime import datetime, timedelta
from pathlib import Path
from prefect import flow, task

@task
def delete_old_local(dir_path: str, days: int):
    cutoff = datetime.now() - timedelta(days=days)
    deleted = 0
    for p in Path(dir_path).rglob("*.parquet"):
        if datetime.fromtimestamp(p.stat().st_mtime) < cutoff:
            p.unlink(missing_ok=True)
            deleted += 1
    return deleted

@flow
def cleanup_local(dir_path: str, days: int):
    deleted = delete_old_local(dir_path, days)
    print(f"Deleted {deleted} old files")

3) Keep results ephemeral - If the Parquet is just a cache, write it to a temp location and delete it at the end of the flow (or avoid persisting large results by returning only small references and keeping

persist_result=False

where possible). How this interacts with Prefect caching - Use

cache_expiration

to limit reuse of cached results (logical expiration), and a lifecycle policy or cleanup flow to remove the physical files. - Consider

cache_result_in_memory=False

for large results to reduce memory pressure. Prefect docs for reference - Results and persistence: Prefect 3.x Results - Caching concepts: Caching If you tell me where you’re storing the Parquet files (local, S3, GCS, Azure) and your retention window, I can share a concrete lifecycle setup or a ready-to-run cleanup flow for that backend.

Great question — you’re right that if you hardcode a single

result_storage_key

(filename), different parameter combinations would overwrite each other. In Prefect 3.x, the standard solution is: - Let Prefect compute a unique cache key per parameter set, and - Either rely on that for deduplication or include parameter info (or a hash of it) in your

result_storage_key

. Here are the common patterns: 1) Easiest: parameter-aware caching with no custom filenames - Use

cache_policy=INPUTS

(or

cache_key_fn=task_input_hash

) to ensure a unique cache key per parameter combination. - Set

persist_result=True

so results are written to disk. - Prefect will look up the previous result by cache key; you don’t need to encode parameters into a filename yourself. Example:

from prefect import task
from prefect.cache_policies import INPUTS

@task(persist_result=True, cache_policy=INPUTS)
def compute(x: int, y: str):
    ...

2) Human-readable or per-parameter filenames - Keep parameter-aware caching as above, and add a

result_storage_key

template that includes parameters or a hash. - The

result_storage_key

is templated with runtime values; you can reference task parameters via runtime context. Example:

from prefect import task
from prefect.cache_policies import INPUTS

@task(
    persist_result=True,
    cache_policy=INPUTS,
    result_storage_key="results/compute/{task_run.parameters.x}-{task_run.parameters.y}.json",
)
def compute(x: int, y: str):
    ...

3) Custom hash in the filename (when parameters are large or not filename-safe) - Define a custom

cache_key_fn

that hashes the

arguments

and reuse that idea in your filename (i.e., include just the hash). - Built-in helper:

prefect.tasks.task_input_hash

. Example:

from prefect import task
from prefect.tasks import task_input_hash

@task(
    persist_result=True,
    cache_key_fn=task_input_hash,
    result_storage_key="results/compute/{task_run.id}.json"  # or include your own hash in the string
)
def compute(x: int, y: str):
    ...

If you want the hash itself in the filename, a common pattern is to: - Use

cache_key_fn

to hash parameters, and - Either put the same hashing logic into your own parameter-to-filename helper or just rely on Prefect’s cache to deduplicate and keep the filename simple (e.g., per-task-run). Notes and tips: - You can exclude specific parameters from the cache key if they shouldn’t affect caching:

cache_policy = INPUTS - "some_param"

. - Avoid putting secrets in filenames — prefer a hash. - Make sure

persist_result=True

(or set the setting that enables persistence by default) if you expect on-disk results. Docs for deeper reference: - Caching concepts: Caching in Prefect 3 - Advanced results (storage keys, persistence): Advanced: Results - Cache policy API: Cache policies API A couple of clarifications to tailor this: - Are you on Prefect 3.x and using local filesystem or cloud storage (S3/GCS/etc.) for results? - Do you want readable filenames with raw parameter values, or is a hash-based filename acceptable?

In Prefect 3.x,

task_input_hash

is a ready-made cache key function you pass to a task’s

cache_key_fn

so each unique set of inputs creates a unique cache key. Basic usage:

from datetime import timedelta
from prefect import flow, task
from prefect.tasks import task_input_hash

@task(
    cache_key_fn=task_input_hash,          # hash all inputs to this task
    cache_expiration=timedelta(hours=1),   # optional expiration
    persist_result=True                    # persist result so it can be reused across runs
)
def add(x: int, y: int) -> int:
    print("Computing (not cached)")
    return x + y

@flow
def my_flow():
    a = add(1, 2)   # computes
    b = add(1, 2)   # served from cache
    c = add(2, 2)   # computes (different inputs)
    return a, b, c

if __name__ == "__main__":
    my_flow()

Notes: -

task_input_hash(context, arguments)

takes the task run context and the dict of bound parameters and returns a deterministic string. If Prefect can’t hash the inputs, it will skip caching for that run. - For most cases,

cache_policy=INPUTS

is an equivalent, simpler alternative. Use

cache_key_fn=task_input_hash

when you explicitly want to control the hashing function. - If you also want per-parameter filenames, add a

result_storage_key

template that includes parameters:

@task(
    cache_key_fn=task_input_hash,
    persist_result=True,
    result_storage_key="results/add/{task_run.parameters.x}-{task_run.parameters.y}.json",
)
def add(x: int, y: int) -> int:
    return x + y

Docs: - Caching concepts: Caching in Prefect 3 - Tasks API (includes

task_input_hash

): prefect.tasks API - Advanced results (storage keys, persistence): Advanced: Results If you share a snippet of your task, I can tailor the

cache_key_fn

and

result_storage_key

to your exact parameter names and storage.

Yes. In Prefect 3.x you can decide at flow runtime whether a task should use cache or bypass it. Common patterns: 1) Toggle caching per call with with_options - Build a “variant” of the task on the fly. Use

NO_CACHE

to disable caching entirely, or

refresh_cache=True

to ignore existing cache and recompute (then update the cache).

from prefect import flow, task
from prefect.cache_policies import NO_CACHE
from prefect.tasks import task_input_hash

@task(cache_key_fn=task_input_hash, persist_result=True)
def expensive(x: int) -> int:
    print("Computing...")
    return x * 2

@flow
def my_flow(x: int, use_cache: bool = True, refresh: bool = False):
    t = expensive
    if refresh:
        # recompute even if cache exists; updates the cache
        t = t.with_options(refresh_cache=True)
    elif not use_cache:
        # disable caching for this call
        t = t.with_options(cache_policy=NO_CACHE, cache_key_fn=None)

    return t(x)

my_flow(10, use_cache=True)   # uses cache if available
my_flow(10, refresh=True)     # forces recompute, updates cache
my_flow(10, use_cache=False)  # no cache read/write

Notes: -

cache_key_fn

takes precedence over

cache_policy

. If the task was defined with a

cache_key_fn

, set it to

None

when you want to disable caching for that call. - Use

persist_result=True

if you want the cached result to be reusable across runs. 2) Make caching conditional inside a custom cache_key_fn - Return

None

from the cache key function to skip caching. You can pass a flag into the task and ignore it in the hash.

from prefect import flow, task
from prefect.tasks import task_input_hash

def conditional_cache_key(context, arguments):
    use_cache = arguments.get("use_cache", True)
    if not use_cache:
        return None   # disables caching for this call
    # Exclude the flag from the hash so it doesn't affect cache identity
    args_wo_flag = {k: v for k, v in arguments.items() if k != "use_cache"}
    return task_input_hash(context, args_wo_flag)

@task(cache_key_fn=conditional_cache_key, persist_result=True)
def expensive(x: int, use_cache: bool = True) -> int:
    print("Computing...")
    return x * 2

@flow
def my_flow(x: int, use_cache: bool):
    return expensive(x, use_cache=use_cache)

3) Quick “refresh but keep caching” per call - If you always use caching but sometimes need a fresh result, just use:

expensive.with_options(refresh_cache=True)(x)

Docs: - Caching overview: Caching in Prefect 3 - Tasks API (includes

with_options

,

task_input_hash

): prefect.tasks API If you share how you’re deciding the flag (flow parameter, env var, etc.), I can suggest the cleanest variant for your case.