Yes — but how depends on what you mean. Short answer: - If you want only the failed task to try again automatically, use task retries. - If the flow run has already failed, Prefect can retry the flow run, but that is a full flow rerun, not a “resume from the failed task” feature. In Prefect 3.x, there is not a built-in “restart this flow from task X” capability. Best option for most cases: put retries on the specific task that may fail. Example:

python
from prefect import flow, task

@task(retries=3, retry_delay_seconds=30)
def flaky_task():
    raise RuntimeError("temporary problem")

@flow
def my_flow():
    flaky_task()

if __name__ == "__main__":
    my_flow()

With this: - the task gets up to 3 retry attempts - successful upstream tasks do not rerun just because this task failed If the whole flow has already ended in a failed state, you can retry the flow run from the CLI:

bash
prefect flow-run retry <FLOW_RUN_ID>

But that reruns the entire flow run, not just the failed task. If your goal is to avoid redoing successful work, common patterns are: - task retries for transient failures - result caching / persistence for expensive successful tasks - designing the flow so tasks are idempotent Relevant docs: - Write tasks - Write flows - Caching - Flow run CLI reference If you want, I can show you the exact Prefect pattern for: 1. retrying one failed task, or 2. rerunning a failed flow without recomputing successful tasks.

Not directly, no. In Prefect 3.x, you cannot resume an existing flow run from the exact failed task after fixing the code or data issue. If you rerun the flow run, Prefect starts the flow again from the beginning. For a 3-hour flow, the usual ways to avoid losing all prior work are: 1. Add retries to the failing task Good if the failure is transient.

python
from prefect import task

@task(retries=3, retry_delay_seconds=60)
def my_task():
    ...

2. Persist/cache outputs of expensive upstream tasks Then when you rerun the flow, earlier successful tasks can be skipped by reusing saved results.

python
from prefect import flow, task
from prefect.cache_policies import INPUTS

@task(cache_policy=INPUTS, persist_result=True)
def expensive_step(x):
    ...
    return x

That way: - first half of the flow completes - one middle task fails - you fix the issue - rerun the flow - cached successful tasks can be reused instead of recomputed 3. Break the flow into smaller flows / stages For example: - flow A: extract - flow B: transform - flow C: load Then if B fails, you rerun B instead of redoing A. For your case, caching/persisted results or splitting the workflow into stages is usually the best answer. A few important caveats: - caching only helps if task inputs/cache keys match - persisted results need storage configured - tasks should be idempotent / safe to rerun Docs: - Caching - Tasks - Flows If you want, I can help you redesign your 3-hour flow so a mid-run failure only requires rerunning the failed portion.