https://outerbounds.com/ logo
Join Slack
Powered by
Hi team, I have a job that uses `with S3(run=self...
# ask-metaflow
c
Hi team, I have a job that uses 
with S3(run=self, prefix=prefix) as s3
to store data on S3 in step A. In Step B, we load it from the same location. Step A succeeded and step B failed. I am then running resume. It looks when I do that, Step B cannot find the previous artifacts (I assume they were stored under the unseccesful run's ID rather than the resumed run's ID). Am I doing something wrong, do I need to use something else than self in cases like this? Or is this expected to work and I might have another issue somewhere?
āœ… 1
c
Hi šŸ‘‹ did you try use 
current.origin_run_id
(docs) to handle this case?
c
Great, didn't know that. So I should probably do something like this?
Copy code
with S3(run=current.origin_run_id if current.origin_run_id else self, prefix=prefix) as
?
c
I think the conditional should work for identifying if the current run is resumed or not, but what you want to pass to the run arg is the metaflow.Run object itself, instead of the id.
šŸ‘ 1
to spell it out, here is a test flow. It will throw an error when you do 
python flow.py run
. Then comment the 
raise ValueError
out, and do 
python flow.py resume
.
Copy code
from metaflow import FlowSpec, step, current, Run, S3

class MyFlow(FlowSpec):

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

    @step
    def middle(self):

        raise ValueError

        if current.origin_run_id is None:
            run = self
        else:
            run_path = "{}/{}".format(self.__class__.__name__, current.origin_run_id)
            run = Run(run_path)

        s3 = S3(run=run)
        ...
        s3.close()

        self.next(self.end)

    @step
    def end(self):
        pass

if __name__ == '__main__':
     MyFlow()
c
That works perfectly, thank you so much!
šŸ’Æ 1
One more question. This works with one resume but at with the second resume, current.origin_run_id points to the first re-attempt instead of the original run id. Any way around that?
c
hmm not sure I follow. when I use the above flow I see the thing that I think you are looking for. namely, the ID of the original run that ran the tasks before the one with the error we resume in. ā€¢ leave ValueError and run --> fail on run ID 1 ā€¢ leave ValueError and resume --> fail on run ID 2 "resumed from 1" ā€¢ leave ValueError and resume --> fail on run ID 3 "resumed from 1" ā—¦ in this flow I print the 
current.origin_run_id
and see ID of flow 1
c
You're right, I was able to reproduce it as follows. Let me check if there's anything else going on.
Copy code
from metaflow import FlowSpec, step, current, Run, S3

class MyFlow(FlowSpec):

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

    @step
    def middle(self):

        print("{}/{}".format(self.__class__.__name__, current.origin_run_id))
        raise ValueError

        self.next(self.end)

    @step
    def end(self):
        pass

if __name__ == '__main__':
     MyFlow()
Copy code
python test_flow.py  run
Metaflow 2.8.1 executing MyFlow for user:jodaiber
Validating your flow...
    The graph looks good!
Running pylint...
    Pylint is happy!
2023-05-19 19:31:35.432 Workflow starting (run-id 6136):
2023-05-19 19:31:37.011 [6136/start/81999 (pid 4140071)] Task is starting.
2023-05-19 19:31:42.769 [6136/start/81999 (pid 4140071)] Task finished successfully.
2023-05-19 19:31:49.305 [6136/middle/82000 (pid 4140133)] Task is starting.
2023-05-19 19:31:52.025 [6136/middle/82000 (pid 4140133)] MyFlow/None
2023-05-19 19:31:52.141 [6136/middle/82000 (pid 4140133)] <flow MyFlow step middle> failed:
2023-05-19 19:31:54.403 [6136/middle/82000 (pid 4140133)] Internal error


python test_flow.py resume
Metaflow 2.8.1 executing MyFlow for user:jodaiber
Validating your flow...
    The graph looks good!
Running pylint...
    Pylint is happy!
2023-05-19 19:32:46.601 Gathering required information to resume run (this may take a bit of time)...
2023-05-19 19:32:56.595 Workflow starting (run-id 6137):
2023-05-19 19:32:57.469 [6137/start/82022] Cloning results of a previously run task 6136/start/81999
2023-05-19 19:33:01.108 [6137/middle/82023 (pid 4140591)] Task is starting.
2023-05-19 19:33:03.722 [6137/middle/82023 (pid 4140591)] MyFlow/6136


python test_flow.py resume
Metaflow 2.8.1 executing MyFlow for user:jodaiber
Validating your flow...
    The graph looks good!
Running pylint...
    Pylint is happy!
2023-05-19 19:33:57.167 Gathering required information to resume run (this may take a bit of time)...
2023-05-19 19:34:03.332 Workflow starting (run-id 6138):
2023-05-19 19:34:04.393 [6138/start/82025] Cloning results of a previously run task 6136/start/81999
2023-05-19 19:34:08.270 [6138/middle/82026 (pid 4140999)] Task is starting.
2023-05-19 19:34:10.804 [6138/middle/82026 (pid 4140999)] MyFlow/6136
8 Views
Open in Slack
PreviousNext
Powered by