Workflow composition, failure handlers, and nodes

Flytekit workflows are composed of tasks, other workflows, and launch plans. While most workflows are defined by passing data between tasks, flytekit also provides low-level control over execution order and node-specific configurations.

Workflow Composition and Promises

In flytekit, calling a task within a @workflow function does not immediately execute the task. Instead, it returns a Promise (defined in flytekit.core.promise.Promise). A Promise is a placeholder for a value that will be computed at runtime.

When you pass the output of one task to another, flytekit automatically creates a data dependency between the underlying nodes.

from flytekit import task, workflow

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

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

@workflow
def my_workflow() -> int:
    # result is a Promise[int]
    result = get_value()
    # Passing the promise to process_value creates a dependency
    return process_value(val=result)

Internally, the Promise object tracks the Node that produces the value. If the task returns multiple values, the Promise can be indexed or accessed via attributes to refer to specific outputs.

Explicit Node Creation

Sometimes you need to define execution order without a direct data dependency. For example, you might want a cleanup task to run only after a processing task finishes, even if the cleanup task doesn't use the processing task's output.

The create_node function in flytekit.core.node_creation allows you to explicitly instantiate a Node. You can then use the >> operator (or the runs_before method) to enforce ordering.

from flytekit import task, workflow
from flytekit.core.node_creation import create_node

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

@task
def work():
    print("Working...")

@workflow
def manual_order_wf():
    setup_node = create_node(setup)
    work_node = create_node(work)
    
    # Enforce that setup runs before work
    setup_node >> work_node

Accessing Outputs from create_node

Unlike a standard task call which returns a Promise directly, create_node returns a Node object (or a VoidPromise if the task has no outputs). To use the outputs of a node created this way, you must access them via the .outputs dictionary or as attributes on the node object.

@task
def produce() -> int:
    return 100

@workflow
def output_access_wf() -> int:
    node = create_node(produce)
    
    # Accessing the output 'o0' (default name for single output)
    # node.outputs["o0"] is a Promise
    return process_value(val=node.outputs["o0"])

Per-Node Overrides

You can customize the execution behavior of individual nodes using the with_overrides method. This is available on both Node objects and Promise objects (which forward the call to their origin node).

Common overrides include:

• Resources: CPU, memory, and GPU limits/requests.
• Retries: Number of times to retry a failed node.
• Timeout: Maximum duration the node is allowed to run.
• Interruptible: Whether the node can run on spot instances.

from flytekit import Resources

@workflow
def override_wf(val: int) -> int:
    promise = process_value(val=val)
    
    # Apply overrides to the node producing this promise
    promise.with_overrides(
        requests=Resources(cpu="2", mem="4Gi"),
        retries=3,
        node_name="custom-process-node"
    )
    return promise

Note: Resource overrides must be static values. You cannot use a Promise to define requests or limits in with_overrides.

Failure Handlers

Flytekit allows you to define a specific task or workflow to run if a workflow fails. This is configured using the on_failure parameter in the @workflow decorator.

A failure handler must accept the same inputs as the workflow itself. It can also optionally accept a FlyteError object (from flytekit.models.core.errors) to inspect the cause of the failure.

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

@task
def cleanup_on_failure(wf_input: str, err: typing.Optional[FlyteError] = None):
    print(f"Workflow failed with input: {wf_input}")
    if err:
        print(f"Error message: {err.message}")

@task
def failing_task(val: str):
    raise ValueError("Something went wrong")

@workflow(on_failure=cleanup_on_failure)
def failure_wf(wf_input: str):
    failing_task(val=wf_input)

When failure_wf fails, Flyte will invoke cleanup_on_failure, passing the original wf_input and the error details. The failure handler's signature must be compatible with the workflow's inputs; any additional parameters in the handler (like err) must be Optional.