from prefect import flow, task
from prefect.filesystems import LocalFileSystem, S3
@flow(result_storage=S3(bucket_path="my-bucket"))
def my_flow():
my_task()
@task(persist_result=True, result_storage_key="my_task.json")
def my_task():
...
my_flow() # The task's result will be persisted to '<s3://my-bucket/my_task.json>'
Paul Reyna
08/14/2023, 10:52 PM
in Prefect 1, we were able to do something like this, where we only had to define it at the flow definition, rather than explicitly defining the S3 key for each task
Copy code
with Flow(
result=S3Result(bucket="MY_S3BUCKET", location=f"{flow_name}/{task_name}/{date}/{task_run_id}")
) as flow:
< flow code >
Paul Reyna
08/14/2023, 10:53 PM
I was wondering if there’s some sort of equivalent in Prefect 2, but it seems like the flow decorator only allows us to pass in the S3 bucket via
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.