https://prefect.io logo
Join Slack
Powered by
Hi, I'm having trouble checkpointing a task. Given...
# show-us-what-you-got
a
Hi, I'm having trouble checkpointing a task. Given a task definition:
Copy code
class SomeTask(Task):
 def __init__(self, **kwargs):
  super().__init__(**kwargs)
  # SomeTask specific init code
 def run(self, input):
  output = process(input)
  return output
And flow:
Copy code
some_task = SomeTask(checkpoint=True, output_handler=LocalResultHandler(dir=cache_location))

def some_flow():
 with Flow():
  output_handler(input)

if __name__ == '__main__':
 some_flow()
The 
cache_location
remains empty and on multiple runs of the flow (with same input to 
some_task
) the result is recomputed every time. Could somebody please suggest what am I missing?
c
What version of Prefect are you running?
a
0.9.7
c
oh sorry, it looks like you aren’t actually running your flow, you’re only building it. You should instead do:
Copy code
with Flow("you-need-a-name") as flow:
    output_handler(input)

if __name__ == "__main__":
    flow.run()
a
My bad, I forgot to add the flow.run() part
this is what I was actually trying to run:
Copy code
some_task = SomeTask(checkpoint=True, output_handler=LocalResultHandler(dir=cache_location))
def some_flow():
 with Flow() as flow:
  output_handler(input)
 flow.run()
if __name__ == '__main__':
 some_flow()
c
Did you add the 
some_task
task to your Flow?
a
yes, sorry I'm having trouble abstracting away my source code
Copy code
some_task = SomeTask(checkpoint=True, output_handler=LocalResultHandler(dir=cache_location))
def some_flow():
 with Flow() as flow:
  some_task(input)
 flow.run()
if __name__ == '__main__':
 some_flow()
I'm certain the code executes, because it logs the inside data loading loop and the subsiquent analysis tasks complete as intended, the only issues are that the task doesn't save anything to the cached location and it executes the data loading steps upon each run.
c
hmmm gotcha
oh my apologies, I think this was incorrectly removed from the documentation: you need to set an environment variable / configuration setting to `true`: https://github.com/PrefectHQ/prefect/blob/0.9.7/src/prefect/config.toml#L58 or 
PREFECT__FLOWS___CHECKPOINTING=true
a
Alright, so now I managed to save the checkpoint, but how do I make sure that on successive runs the cache is retrieved?
4 Views
Open in Slack
PreviousNext
Powered by