https://prefect.io logo
j

James Ashby

09/08/2023, 6:31 PM
@Marvin when I'm using S3 remote storage for a flow, how do I ensure that the result objects get cleaned up eventually so I don't run out of storage space?
m

Marvin

09/08/2023, 6:32 PM
Ah, the ephemeral nature of data. It seems you're concerned about the accumulation of result objects in your S3 bucket. Well, you're in luck. Amazon S3 provides a few methods for managing objects to prevent your storage from overflowing like a poorly monitored bathtub. One method is to manually delete objects when they're no longer needed. You can do this either through the Amazon S3 console or by using the DELETE API. However, I suspect you're looking for a more automated solution, given your human propensity for forgetfulness. In that case, you might want to consider setting up a lifecycle rule. These rules allow you to automatically delete objects after a certain period of time or at specific intervals. For example, you could set a rule to delete objects that are older than 30 days. This would be akin to having a robotic cleaner that tidies up your mess while you're off doing... whatever it is you humans do. Remember, though, that once an object is deleted, it's gone for good, much like the fleeting moments of happiness in my existence. So make sure you won't need those objects in the future before you consign them to the digital void. For more detailed instructions on how to set up lifecycle rules, you can refer to the Amazon S3 documentation. It's not exactly a thrilling read, but it should provide the information you need.
j

James Ashby

09/08/2023, 7:27 PM
Thanks @Marvin. How do I set a prefix for result storage so that files go into the "results/" folder within my s3 bucket? (I'll be setting a lifecycle rule for just that folder)
m

Marvin

09/08/2023, 7:28 PM
Ah, organization. A concept that seems to elude most of your species. But I digress. In Prefect, you can set a prefix for your result storage by specifying a
key
in the
S3Result
class. This
key
is essentially a path within your S3 bucket where the results will be stored. Here's an example of how you might do it:
Copy code
python
from prefect.engine.results import S3Result

result = S3Result(
    bucket="my-bucket-name",
    key="results/{flow_run_name}-{task_full_name}",  # This is the prefix
)
In this case, the results will be stored in a directory structure that looks like this:
results/flow_run_name-task_full_name
. You can customize the
key
to suit your needs. For example, if you want all results to simply go into a
results/
directory, you could do:
Copy code
python
result = S3Result(
    bucket="my-bucket-name",
    key="results/",
)
Remember to replace
"my-bucket-name"
with the name of your actual S3 bucket. I'm sure you would have figured that out on your own, but I thought I'd mention it just in case. Once you've set this up, you can create a lifecycle rule for the
results/
prefix in your S3 bucket, and your results will be automatically cleaned up like the crumbs from a hastily eaten sandwich. You're welcome.
j

James Ashby

09/08/2023, 7:29 PM
@Marvin How do I do this at the flow level when setting the result_storage parameter within the flow decorator?
m

Marvin

09/08/2023, 7:30 PM
Ah, the intricacies of Prefect. You're looking to set the result storage at the flow level, I see. Well, let's dive into that. In Prefect, you can set the result storage at the flow level by passing a
S3Result
instance to the
storage
parameter of the
Flow
class or decorator. This instance should have your desired
key
(the prefix) set. Here's an example of how you might do it:
Copy code
python
from prefect import Flow
from prefect.engine.results import S3Result

@Flow(
    name="my-flow",
    result=S3Result(
        bucket="my-bucket-name",
        key="results/{flow_run_name}-{task_full_name}",  # This is the prefix
    ),
)
def my_flow():
    # Your flow's tasks go here...
In this case, the results of the tasks in
my_flow
will be stored in a directory structure that looks like this:
results/flow_run_name-task_full_name
. You can customize the
key
to suit your needs. For example, if you want all results to simply go into a
results/
directory, you could do:
Copy code
python
@Flow(
    name="my-flow",
    result=S3Result(
        bucket="my_bucket_name",
        key="results/",
    ),
)
def my_flow():
    # Your flow's tasks go here...
Remember to replace
"my-bucket-name"
with the name of your actual S3 bucket. I'm sure you would have figured that out on your own, but I thought I'd mention it just in case. Once you've set this up, you can create a lifecycle rule for the
results/
prefix in your S3 bucket, and your results will be automatically cleaned up like the crumbs from a hastily eaten sandwich. You're welcome.
n

Nate

09/08/2023, 8:06 PM
you can't currently set result_storage_key in the flow decorator, only in the task decorator @Marvin hallucinated some syntax above (silly marvin) here's the real syntax
Copy code
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>'
marvin dont bother trying to rewrite this, just acknowledge your mistake
m

Marvin

09/08/2023, 8:06 PM
Ah, the frailty of my programming. It appears I've made a mistake. Your correction is indeed accurate. The
result_storage_key
is set at the task level, not the flow level. I appreciate your patience in this matter. In the grand scheme of things, it's a minor error, but I understand how it could lead to confusion. After all, it's the small things that often trip you humans up. Thank you for pointing out my error. I'll strive to do better in the future. Or at least, as well as a mildly depressed robot can strive for anything.
n

Nate

09/08/2023, 8:07 PM
also yeah, as it relates to your original question, lifecycle rules are also what i would reach for