lucian
06/29/2025, 5:01 AMMarvin
06/29/2025, 5:01 AMMarvin
06/29/2025, 5:02 AMpython
from prefect import flow, task
from prefect.tasks.core.defaults import cache_policy
@task(cache_policy=cache_policy(
key_storage="/path/to/cache/directory" # This acts as your cache prefix
))
def my_cached_task():
return "some result"
2. Using a remote storage block (like S3):
python
from prefect import flow, task
from prefect.tasks.core.defaults import cache_policy
from prefect_aws import S3Bucket
# First, ensure you have configured your S3 block
s3_block = S3Bucket.load("my-block-name")
@task(cache_policy=cache_policy(
key_storage=s3_block,
prefix="my-custom-prefix/" # Optional prefix within the storage location
))
def my_cached_task():
return "some result"
The cache key will be automatically generated based on:
- The task's inputs
- The task's code definition
- The flow run or task run ID
These components are hashed together to create a unique cache key, and the prefix/storage location you specify determines where these cache records are stored.
Remember that caching is particularly useful for:
- Expensive computations that you want to avoid repeating
- Tasks that frequently produce the same results for the same inputs
- Optimizing workflow execution time
For more details on caching configuration and best practices, you can refer to the Prefect caching documentation.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