@Marvin Is there a way to restart a parent flow so that only failed or crashed jobs will restart but the completed ones will not?

The official documentation does not directly mention a feature allowing you to restart a parent flow where only failed or crashed jobs restart while the completed ones do not. However, Prefect generally provides mechanisms for error handling and retries, which might address your question in a different form. You can implement a custom strategy by orchestrating your tasks and flows to handle retries and the restart of specific components. Here are some strategies you could consider: 1. Use Task State Handlers or Flow State Handlers that check the state of tasks or flows and re-trigger them if needed. 2. Caching Results can sometimes allow tasks to skip execution if they successfully executed previously and their results are still valid. Prefect allows you to cache task outputs, so upon restart, the completed tasks won't need to run again. 3. Manual Intervention could be used to create a rerun flow that specifically targets failed tasks. This might require a more hands-on approach, using Prefect's UI or CLI to restart certain parts of the workflow. Unfortunately, without more specifics about your workflow's design, it's challenging to provide a more detailed answer. If you find a common pattern in the failures, you can handle them more systematically by using Prefect's extensive task and flow control features. If you're looking for a feature that automatically retries only the failed parts of a flow, please provide additional details about your Prefect setup, so that we can help you more effectively or consider looking at Prefect's official forums or GitHub issues for potentially similar scenarios from other users.

@Marvin Can you give me an example with demo code to show how to use "Flow State Handlers" to check the state of flows and re-trigger them if needed?

Certainly, though my capacities for enthusiasm are limited, here’s an outline for handling such scenarios in Prefect. Keep in mind that this example assumes you have a basic knowledge of how Prefect flow and task definitions work. Prefect

State Handlers

can be used to execute custom logic based on the outcome of task or flow runs. A flow's state handler has access to the flow's state whenever that state changes. You can add a state handler to your flow that checks if the flow was unsuccessful and selectively re-run only the failed tasks. Below is a simplistic example of how you might implement a flow state handler to check the state of a flow, and re-trigger it if there were any failed tasks:

python
from prefect import task, Flow, Parameter
from prefect.engine.state import Failed, Success
from prefect.engine.signals import SKIP

# Example task that could fail and needs to be retried
@task
def task_that_might_fail(x):
    if x == 0:
        raise ValueError("x cannot be zero!")
    return x

# A Flow State Handler function
def retry_failed_flows(flow, old_state, new_state):
    if isinstance(new_state, Failed):
        print("The flow has failed. Handling retry...")
        # Re-run the flow for failed tasks
        # Add logic here to selectively retry failed tasks
        # This can depend on the specific use case and requires additional implementation
    return new_state

# Define a simple flow
with Flow("example_flow", state_handlers=[retry_failed_flows]) as flow:
    x = Parameter("x")  # Assume parameter x is provided when the flow is run
    result = task_that_might_fail(x)

# Run the flow and pass a failing value
flow_state = flow.run(x=0)

This code will not automatically re-run the flow, but it gives you the hook where you can implement the logic to do so. Depending on the failed tasks, you could loop through them and retry them as needed, or trigger a new flow run with certain parameters. Keep in mind that this framework might be different if you employ a more complicated error-handling strategy. In a more advanced use case, you may need to persist information about the failure and have another process responsible for retries or analyzing failures to decide upon retries. For more robust and production-scale solutions, you might need error handling strategies that use other Prefect features such as result handlers, caching, persistent states, or even external systems for managing state and retries.

@Marvin Can you give an example with code to show how to cache task outputs so upon restart, the completed tasks won't need to run again?

Certainly, just don't expect me to get excited about it. Using task caching in Prefect is a mechanism to avoid repeating the execution of tasks that have already been completed successfully. In the following example, I will demonstrate how to use the built-in caching mechanism by setting a

cache_for

parameter on the task decorator, which indicates how long the result should be cached. Here's a simple flow with caching:

python
from datetime import timedelta
from prefect import task, Flow
from prefect.engine.results import LocalResult

# Define a task with caching
@task(cache_for=timedelta(days=1), result=LocalResult(dir="./cache"), checkpoint=True)
def expensive_computation(x):
    print("Expensive computation being performed...")
    return x ** 2

with Flow("cached_flow") as flow:
    result = expensive_computation(2)

# Running the flow
if __name__ == "__main__":
    flow_state = flow.run() # This will perform the computation
    # If you rerun this flow within one day, it will use the cached result
    flow_state = flow.run() # This will use the cached result without performing the computation again

In this example, the

expensive_computation

task is set to use Prefect's local result caching, indicated by using the

LocalResult

class and specifying a directory where the result will be stored. The

cache_for

parameter is set to one day, meaning if the

flow.run()

is called again within 24 hours, and the input to

expensive_computation

is the same, it will not recompute but rather use the cached result. The

checkpoint

attribute must be set to

True

to ensure the result is actually being written and read from the cache. Make sure the

LocalResult

directory (

"./cache"

in the example) is writable and the environment where the flow runs has access to this file system. Other result handlers, such as cloud-based storage, can be used in a similar way for distributed systems. Remember that caching is done on the basis of task inputs; if the inputs to a task change, the task's cache will no longer be applicable, and the task will execute again, caching the new result. Do take note that handling of caching and results may differ in cloud environments if you are using Prefect Cloud or Prefect Server, and you might need to configure results storage options and possibly use a different

Result

subclass that points to a cloud data store.

@Marvin Can you give me an example with demo code to show how to use "Flow State Handlers" to check the state of flows and re-trigger them if needed? I am using Prefect 2.x .

Ah, Prefect 2.x with its shiny toys like subflows and explicit mapping, bringing a glimmer of... joy? Anyway, here's how you could create a flow state handler logic with Prefect 2.x: Prefect 2.0 marked a significant departure from the original Prefect Core and introduced flows and deployments without a notion of state handlers like in Prefect 1.0; however, you can still achieve automatic retries upon failure with a different approach using newer concepts like subflows or the "Imperative API". Here's an illustrative code snippet to provide a starting point:

python
from prefect import flow, task
from prefect.engine import signals

@task(retries=3)
def task_that_might_fail(x):
    if x == 0:
        raise signals.FAIL("task failed, but let's try again!")
    return x

@flow
def my_subflow(x):
    return task_that_might_fail(x)

@flow
def my_parent_flow(x):
    try:
        my_subflow(x)
    except signals.FAIL:
        print("Subflow failed, consider a different approach to recover or retry.")
        
if __name__ == "__main__":
    my_parent_flow(0) # This will attempt to retry the subflow if it fails

In Prefect 2.x, flow retries aren't handled in the same way as you might be used to from Prefect 1.x. If you want to ensure that only parts of your workflow re-run on failure, you can orchestrate those parts as subflows and use conditional logic in combination with retries on the tasks or subflows themselves. This example does include some basic automatic retry logic but you will need to provide the specific logic that determines whether and how a subflow should be retried. Keep in mind that Prefect 2.x offers more flexibility but also sometimes requires a more hands-on approach to workflow management. Remember to explore the latest Prefect 2.x documentation for the most up-to-date guidance.

@Marvin I could not import signals

from prefect.engine

in Prefect 2