Workflow composition and nodes

Flyte workflows are defined as a collection of nodes connected by data dependencies or explicit execution orders. In flytekit, the @workflow decorator transforms a Python function into a declarative graph where each task call or sub-workflow invocation creates a Node.

Defining Workflows with the @workflow Decorator

The @workflow decorator captures the structure of your computation. When you call a task inside a workflow, flytekit does not execute the task immediately. Instead, it records the call as a Node in the workflow graph.

from flytekit import task, workflow

@task
def say_hello(name: str) -> str:
    return f"Hello, {name}!"

@workflow
def my_workflow(name: str) -> str:
    # This call creates a Node in the graph
    greeting = say_hello(name=name)
    return greeting

Inside the workflow function, the variables you work with (like greeting above) are not actual strings or integers; they are Promise objects. These promises represent future values that will be produced when the workflow runs on the Flyte cluster.

Understanding Nodes

A Node (implemented in flytekit.core.node.Node) is the fundamental unit of execution within a workflow. Every time you call a task, sub-workflow, or launch plan inside a @workflow function, flytekit creates a node to represent that step.

A node encapsulates:

• Inputs: Bindings that connect workflow inputs or other node outputs to this node.
• Metadata: Information like retries, timeouts, and interruptibility.
• Flyte Entity: The actual task or sub-workflow to be executed.

Explicit Node Creation

While calling tasks directly is the most common pattern, you can use create_node from flytekit.core.node_creation to explicitly instantiate a node. This is particularly useful when you need to manage tasks that do not have data dependencies.

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

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

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

@workflow
def manual_node_wf():
    # Create nodes explicitly
    setup_node = create_node(setup)
    work_node = create_node(do_work)
    
    # Define execution order
    setup_node >> work_node

Connecting Nodes

Nodes are connected in two primary ways: data dependencies and explicit ordering.

Data Dependencies (Implicit)

When the output of one task is passed as an input to another, flytekit automatically creates a dependency between the corresponding nodes. The downstream node will not execute until the upstream node completes and its outputs are available.

Explicit Ordering with >>

If two tasks do not share data but must run in a specific order (e.g., a setup task must finish before a processing task starts), use the right-shift operator >>. This operator is syntactic sugar for the runs_before method on the Node class.

@workflow
def ordered_wf():
    t1_node = create_node(task_1)
    t2_node = create_node(task_2)
    
    # task_1 will finish before task_2 starts
    t1_node >> t2_node 

Internally, node_1 >> node_2 calls node_1.runs_before(node_2), which appends node_1 to the _upstream_nodes list of node_2.

Customizing Nodes with Overrides

You can customize the execution behavior of a specific node using the with_overrides method. This allows you to set resource requirements, retries, and timeouts for a single instance of a task without changing the task definition itself.

@workflow
def override_wf(val: int):
    # Override resources and retries for this specific node
    node = create_node(my_task, val=val).with_overrides(
        requests=Resources(cpu="2", mem="4Gi"),
        retries=3,
        timeout=3600,
        node_name="custom-node-name"
    )

The with_overrides method modifies the Node instance's internal _metadata and _resources. Note that these overrides must be static values; you cannot use a Promise from another task to set a timeout or resource limit.

Workflow-Level Configuration

The @workflow decorator accepts several parameters to control the behavior of the entire graph:

• on_failure: Specifies a task or sub-workflow to run if the workflow fails. This is often used for cleanup operations.
• failure_policy: Determines how the workflow behaves when a node fails. Options include WorkflowFailurePolicy.FAIL_IMMEDIATELY or WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE.
• interruptible: A boolean indicating if the tasks within the workflow can be scheduled on interruptible (spot) instances.

from flytekit import workflow, WorkflowFailurePolicy

@workflow(
    on_failure=cleanup_task,
    failure_policy=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE,
    interruptible=True
)
def robust_workflow(data: str):
    ...

Declarative Execution Gotchas

Because the code inside a @workflow function is used to build a static graph, standard Python logic often doesn't behave as expected:

1. No Python Conditionals: You cannot use if x == 5: inside a workflow if x is a Promise. You must use Flyte's conditional construct.
2. No Immediate Values: You cannot print the result of a task call inside the workflow function. print(task_call()) will print a Promise object, not the task's return value.
3. Static Overrides: Parameters passed to with_overrides (like cpu or mem) must be known at compile time. They cannot be calculated by an upstream task in the same workflow.