Scott Brownlie01/14/2020, 9:22 PM
does a similar thing, however I can't seem to get it to work as I expected. I have implemented the following toy example:
I would expect the output from each task to be saved to the specified directory automatically but it's not. Is that not what is supposed to happen?
from prefect.engine.result_handlers import LocalResultHandler from prefect import task, Task, Flow results_dir = 'prefect_results/' @task(checkpoint=True) def set_input(): return 10 @task(checkpoint=True) def square(x): return x**2 with Flow("test", result_handler=LocalResultHandler(dir=results_dir)) as flow: task1 = set_input() task2 = square(task1) flow.run()
Chris White01/14/2020, 9:31 PM
it will work as you expect; this behavior will change when we release 0.9.0 so that the default behavior is to checkpoint anytime a result handler is present!
Scott Brownlie01/15/2020, 12:23 PM
Chris White01/15/2020, 4:12 PM
Scott Brownlie01/15/2020, 4:25 PM
Chris White01/15/2020, 4:48 PM
cache validator. Next, provide the values you chose above to the initialization of each of the pre-processing tasks as described in the above doc. On each run, they will now look at the available cached states (which are stored in memory if you run with Prefect Core alone) and decide whether a valid cached state is available or not. If one is available, the task will not rerun and instead return the value from its most previous cached run. For a silly but easy example, check out: https://docs.prefect.io/core/examples/cached_task.html