How can I restart a flow in Prefect 2.0?

Archive
, , ,
8

:100: yes - this is also best practice in data engineering to ensure your components are as self-contained and dependency-free as possible. Doing such check can even be something as simple as querying theMAX(LAST_UPDATED) timestamp field in your database, if you have that

Another alternative is to leverage caching - assuming you run this every hour, if you would add caching with 1 h expiration, this will have exactly the effect you are looking for: successful tasks won’t be rerun, but those that failed and haven’t been cached, will run again in the subsequent run.

from datetime import timedelta
from prefect import flow, task
from prefect.tasks import task_input_hash

@task(cache_key_fn=task_input_hash, cache_expiration=timedelta(hours=1))
def hello_task(name_input):
    # Doing some work
    print("Saying hello")
    return "hello " + name_input

@flow
def hello_flow(name_input):
    hello_task(name_input)