Workflow composition, failure handlers, and nodes

Flytekit workflows are defined using the @workflow decorator, which orchestrates the execution of tasks and sub-workflows. While data flow (passing outputs of one task to the inputs of another) implicitly defines the execution graph, flytekit provides explicit mechanisms for node management, dependency control, and failure handling.

Workflow Composition and Promises

In flytekit, calling a task inside a workflow does not immediately execute the task. Instead, it returns a Promise (or a VoidPromise for tasks with no return values). These promises represent future values that will be available during execution.

When you pass a Promise from one task to another, flytekit automatically creates a dependency between the underlying nodes in the workflow graph.

from flytekit import task, workflow

@task
def get_data() -> int:
    return 42

@task
def process_data(val: int) -> int:
    return val + 1

@workflow
def my_workflow() -> int:
    # data_promise is a Promise object
    data_promise = get_data()
    # Passing the promise creates a dependency: get_data -> process_data
    return process_data(val=data_promise)

Accessing Promise Attributes

If a task returns a complex type like a dataclass or a dict, you can access specific attributes or keys on the Promise. Flytekit records these access paths in the Promise._attr_path and resolves them at runtime.

@task
def get_map() -> dict:
    return {"a": 1, "b": 2}

@workflow
def attr_workflow() -> int:
    m = get_map()
    # Accessing a key on a promise returns a new Promise with an updated attr_path
    return process_data(val=m["a"])

Explicit Node Creation

While implicit dependencies are standard, the create_node function in flytekit.core.node_creation allows you to explicitly instantiate a Node. This is particularly useful for defining execution order when there is no data dependency between tasks.

Defining Dependencies with >>

You can use the right-shift operator >> (implemented via Node.__rshift__) or the runs_before method to force a specific execution order.

from flytekit import task, workflow, create_node

@task
def setup():
    print("Setting up...")

@task
def cleanup():
    print("Cleaning up...")

@workflow
def manual_node_wf():
    setup_node = create_node(setup)
    cleanup_node = create_node(cleanup)
    
    # Ensure setup runs before cleanup even though no data is shared
    setup_node >> cleanup_node

Accessing Outputs from Nodes

Unlike a direct task call which returns a Promise, create_node returns a Node object. The outputs of the task are attached to this node as attributes (e.g., .o0, .o1) and are also available in the node.outputs dictionary.

@task
def produce_multiple() -> (int, str):
    return 1, "hello"

@workflow
def node_output_wf():
    n = create_node(produce_multiple)
    
    # Accessing outputs from the Node object
    val_int = n.o0
    val_str = n.outputs["o1"]
    
    return val_int

Per-Node Overrides

The Node class provides a with_overrides method that allows you to customize the execution parameters of a specific task instance within a workflow. You can also call with_overrides directly on a Promise, which forwards the configuration to the underlying node.

Common overrides include:

• requests and limits: Resource specifications using flytekit.Resources.
• timeout: A datetime.timedelta or integer seconds.
• retries: Number of retry attempts.
• interruptible: Boolean for spot instance usage.

from flytekit import Resources

@workflow
def override_wf(val: int) -> int:
    # Applying overrides to a specific task call
    return process_data(val=val).with_overrides(
        requests=Resources(cpu="2", mem="200Mi"),
        retries=3,
        timeout=600
    )

Failure Handlers

The @workflow decorator supports an on_failure parameter to handle workflow-level failures. This is typically used for cleanup tasks or notifications.

Signature Requirements

A failure handler must be a Flyte task. It must accept all inputs that the workflow itself accepts. Additionally, it can accept an optional err argument of type str to receive the error message that caused the failure. Any other arguments in the failure handler must be optional.

If the signatures do not match, flytekit raises a FlyteFailureNodeInputMismatchException.

@task
def clean_up(name: str, err: str):
    print(f"Workflow for {name} failed with error: {err}")

@task
def failing_task(name: str):
    raise ValueError(f"Failure in task for {name}")

@workflow(on_failure=clean_up)
def failure_wf(name: str):
    # If failing_task fails, clean_up(name=name, err=...) is automatically invoked
    failing_task(name=name)

In this example, clean_up matches the workflow's input name and includes the optional err string, satisfying the requirement for failure handlers.