I have a flow that has 2 steps. The first steps does some compute expensive work and writes the results to S3. The second step does work on the files written by Step 1. I would like to use the

resume

feature so as to not re-run Step 1 again while I tinker and play around with Step 2. However when I try to resume a failed flow and access the file in Step using

input_file = S3(run=self).get("my_file_from_step1.json")

I get an error saying no such file “my_file_from_step1.json” because in this new run there isn’t this file. But I would like to access this file from the successful Step 1 task which belonged to a previous run. Any suggestions on how to do this in a clean way without a bunch of if/else statements for various success/failure scenarios.

Here is a small minimalistic example

from metaflow import FlowSpec, step, S3, current, Run
class S3ResumeExampleFlow(FlowSpec):

@step
    def start(self):
        self.next(self.save_run_id)

@step
    def save_run_id(self):
        self.run_id = current.run_id
        self.next(self.train)
    
    @step
    def train(self):
        # This ensures that when you resume the 
        # `train` step, the run-id will be from the previous run
        with S3(run=Run(self.run_id)) as s3:
            s3.get("data.txt", "data.txt")
        self.next(self.end)

@step
    def end(self):
        pass

if __name__ == '__main__':
    S3ResumeExampleFlow()

Awesome. The only thing I would need to remember is that file will be written to the directory of the first run in S3 even though it is marked as failed in Metaflow. (Unless I write the file to S3 in the resumed run as well). In any case, this is much better than what I was planning to do. Thank you!