Prefect workflow: How to persist data of previous/every schedule run?

user_11853415 0
Viewed 528

In prefect workflow, I'm trying to persist data of every schedule run. I need to compare data of every previous and current result. I tried Localresult and checkpoint=true but its not working. For example,

from prefect import Flow, task
from prefect.engine.results import LocalResult
from prefect.schedules import IntervalSchedule
from datetime import timedelta, datetime
import os
import prefect

@task("func_task_target.txt", checkpoint=True, result=LocalResult(dir="~/.prefect"))
def file_scan():
    files = os.listdir(test) 
    #prefect.context.a = files
    return files

schedule = IntervalSchedule(interval=timedelta(seconds=61))

with Flow("Test persist data", schedule) as flow:
    a = file_scan()
flow.run()

My flow scheduled for every 61 seconds/a minute. On the first run I might get empty result but for the 2nd scheduled run I should get previous flow result to compare. can anyone help me to achieve this? Thanks!

prefect python
1 Answers

Update (15 November 2021) :

Not sure what is the reason, LocalResult and checkpoint actually worked when I ran the registered flow through the dashboard or cli prefect run -n "your-workflow.py" --watch. It doesn't work when I manually trigger the flow (e.g.: flow.run) in python code.

Try these following two options:

Option 1 : using target argument:

https://docs.prefect.io/core/concepts/persistence.html#output-caching-based-on-a-file-target

@task(target="func_task_target.txt", checkpoint=True, result=LocalResult(dir="~/.prefect"))
def func_task():
    return "999"

Option 2 : instantiate LocalResult instance and invoke write manually.

MY_RESULTS = LocalResult(dir="./.prefect"). 

@task(checkpoint=True, result=LocalResult(dir="./.prefect"))
def func_task():
    MY_RESULTS.write("999")
    return "999"

PS:

Having same problem as LocalResult doesn't seem to work for mewhen used in decorator . e.g :

@task("func_task_target.txt", checkpoint=True, result=LocalResult(dir="~/.prefect"))
def file_scan():
user_49271720
Related