Workflow composition, failure handlers, and nodes

Flytekit workflows are defined using the @workflow decorator, which transforms a Python function into a directed acyclic graph (DAG) of execution units called nodes. While the syntax resembles standard Python, flytekit uses a system of Promises and Nodes to orchestrate execution, manage data dependencies, and handle failures.

Workflow Composition and Promises

When you call a task inside a workflow, it does not return the actual value (like an int or str). Instead, it returns a flytekit.core.promise.Promise. This object acts as a placeholder for a future value that will be computed by the Flyte engine.

from flytekit import task, workflow

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

@task
def process_id(id_val: int) -> str:
    return f"ID-{id_val}"

@workflow
def my_wf() -> str:
    # id_promise is a Promise object, not an int
    id_promise = get_id()
    # Passing the promise to another task creates a data dependency
    return process_id(id_val=id_promise)

Internally, the Promise class (found in flytekit/core/promise.py) tracks the origin node and the specific output variable it represents. If a task returns multiple values, flytekit returns a tuple of Promise objects. If a task returns nothing, it returns a VoidPromise.

Logical Operations on Promises

Because Promise objects are not actual values during workflow compilation, you cannot use standard Python truth value testing (e.g., if my_promise:). Doing so raises a ValueError. For logical operations within conditionals, flytekit provides bitwise operator overrides (& for AND, | for OR) and comparison methods like .is_true() or .is_false().

Explicit Node Creation

While calling tasks directly is the standard way to build workflows, you can use create_node from flytekit.core.node_creation for more granular control. This is particularly useful for:

1. Non-data dependencies: Forcing one task to run before another when they don't share data.
2. Per-node overrides: Applying specific resource limits or retries to a single instance of a task.

Accessing Outputs from create_node

A critical distinction exists between calling a task and using create_node. When you call a task, you get a Promise. When you use create_node, you get a Node object (or a VoidPromise). To access the outputs of a node, you must use the .o0, .o1, etc., attributes or the .outputs dictionary.

from flytekit.core.node_creation import create_node

@workflow
def imperative_wf(val: int) -> str:
    # Create nodes explicitly
    node_a = create_node(get_id)
    
    # Access output via .o0 (the first output)
    node_b = create_node(process_id, id_val=node_a.o0)
    
    # Set execution order without data dependency
    node_a >> node_b  # node_a.runs_before(node_b)
    
    return node_b.o0

Per-Node Overrides

The Node class in flytekit/core/node.py provides a with_overrides method. This allows you to customize the execution environment for a specific node without changing the task definition itself. You can override:

• Resources: requests and limits using flytekit.Resources.
• Metadata: timeout, retries, and interruptible status.
• Container Image: Use container_image to run a specific node in a different Docker image.

from flytekit import Resources

@workflow
def override_wf(val: int):
    # Apply overrides to a task call
    t1_node = process_id(id_val=val).with_overrides(
        node_name="heavy-processor",
        requests=Resources(cpu="2", mem="500Mi"),
        retries=3,
        timeout=3600
    )

Failure Handlers

Flytekit allows you to define a cleanup or notification task that runs if a workflow fails. This is configured via the on_failure parameter in the @workflow decorator.

Signature Requirements

The on_failure handler must be a task or another workflow. Its signature must satisfy two conditions:

1. It must accept all inputs defined in the parent workflow.
2. It can optionally accept a FlyteError object (from flytekit.models.core.errors) to inspect the failure details. This parameter must be named err or error and should be Optional.

import typing
from flytekit import task, workflow
from flytekit.models.core.errors import FlyteError

@task
def clean_up(name: str, err: typing.Optional[FlyteError] = None):
    if err:
        print(f"Workflow failed for {name} with error: {err.message}")
    else:
        print(f"Workflow failed for {name} due to an unknown error")

@workflow(on_failure=clean_up)
def cluster_wf(name: str):
    # If any task here fails, clean_up(name=name, err=...) is invoked
    t1(name=name)

When a failure occurs, Flyte automatically captures the inputs passed to the workflow and the error that caused the crash, passing them to the on_failure entity. This ensures that resources can be released or stakeholders notified even when the main logic fails.