Workflow composition and nodes

Flyte workflows are declarative structures that define a directed acyclic graph (DAG) of execution. In flytekit, this graph is composed of Nodes, where each node represents an instance of a task, a sub-workflow, or a launch plan.

Workflow Compilation and Nodes

When you decorate a function with @workflow, flytekit does not execute the function body at runtime on the Flyte platform. Instead, the function is evaluated once at compilation time (typically during registration) to build the workflow topology.

Inside the workflow body, every call to a Flyte entity (like a @task) creates a Node. These nodes are connected via Promises, which represent the future outputs of a node.

The Node Class

The flytekit.core.node.Node class is the fundamental building block of a workflow graph. It tracks:

• ID: A unique identifier within the workflow (e.g., n0, n1), often dnsified via _dnsify to ensure Kubernetes compatibility.
• Flyte Entity: The underlying task, workflow, or launch plan the node executes.
• Bindings: How the inputs of this node are mapped to outputs of upstream nodes or workflow inputs.
• Upstream Nodes: A list of nodes that must complete before this node can start.

Automatic Node Creation

In a standard @workflow, nodes are created automatically when you call a task. flytekit uses a CompilationState to track these nodes as they are defined.

@workflow
def my_workflow(a: int) -> int:
    # Calling add_5 creates a Node and returns a Promise
    x = add_5(a=a) 
    # Passing the Promise 'x' to another task creates a dependency
    return add_5(a=x)

Internally, flytekit.core.promise.flyte_entity_call_handler intercepts these calls. If a compilation context exists, it invokes create_and_link_node, which instantiates the Node, calculates upstream_nodes by inspecting the input Promise objects, and registers the node in the current CompilationState.

Explicit Composition with create_node

While automatic wiring is standard, you can use flytekit.core.node_creation.create_node for explicit control. This is useful when you need to reference a node directly or handle tasks that do not return values.

Manual Dependency Ordering

If two tasks do not share data but must run in a specific order, you can use the >> operator or the runs_before method on the Node object.

from flytekit.core.node_creation import create_node

@workflow
def ordering_wf():
    node_a = create_node(task_a)
    node_b = create_node(task_b)
    
    # node_a must finish before node_b starts
    node_a >> node_b 
    # Equivalent to: node_a.runs_before(node_b)

Accessing Node Outputs

When using create_node, you access outputs via the .outputs property or shorthand attributes like .o0, .o1.

@workflow
def explicit_wf(a: str) -> str:
    t1_node = create_node(t1, a=a)
    return t1_node.o0  # Access the first output of the node

Node Overrides

You can modify the behavior of a specific node without changing the underlying task definition using with_overrides. This method returns the Node (or a Promise wrapping it) with updated metadata.

@workflow
def override_wf(a: int) -> int:
    promise = add_5(a=a).with_overrides(
        node_name="custom-node-name",
        requests=Resources(cpu="1", mem="200Mi"),
        limits=Resources(cpu="2", mem="500Mi"),
        retries=3,
        timeout=datetime.timedelta(minutes=5),
        interruptible=True
    )
    return promise

Key Override Behaviors

• Resource Clamping: If you provide requests without limits, flytekit logs a warning and clamps requests to the original task limits. You cannot provide both a Resources object and individual requests/limits arguments simultaneously.
• DNS Compliance: The node_name override is passed through _dnsify, converting names like my_task_node to my-task-node.
• Caching: You can override caching behavior by passing a Cache object. Note that deprecated parameters like cache_version or cache_serialize will raise a ValueError if used alongside a Cache object.

Imperative Workflows

For scenarios where a workflow's structure is determined dynamically (e.g., based on a configuration file), flytekit provides the ImperativeWorkflow class. Unlike the @workflow decorator, this class allows you to build the DAG programmatically.

from flytekit.core.workflow import ImperativeWorkflow

wb = ImperativeWorkflow(name="dynamic_workflow")
# Define inputs
wf_input = wb.add_workflow_input("in1", str)
# Add nodes
node = wb.add_entity(t1, a=wf_input)
# Define outputs
wb.add_workflow_output("out1", node.outputs["o0"])

ImperativeWorkflow maintains its own CompilationState and validates that all inputs are bound before the workflow is considered ready().

Workflow Metadata and Failure Handling

The @workflow decorator accepts several parameters that govern the entire execution graph:

• failure_policy: Uses WorkflowFailurePolicy. FAIL_IMMEDIATELY (default) stops the workflow on any node failure, while FAIL_AFTER_EXECUTABLE_NODES_COMPLETE allows independent nodes to finish.
• on_failure: Specifies a task or workflow to run if the workflow fails. This node is assigned the ID efn (defined in flytekit.core._common_constants.DEFAULT_FAILURE_NODE_ID).
• interruptible: Sets the default interruptible state for all nodes in the workflow.

When PythonFunctionWorkflow.compile() runs, it validates the on_failure handler by ensuring its interface matches the workflow's inputs. Any additional inputs required by the failure handler must be Optional.

Serialization to Executable Definitions

The final step of composition is translating the Python Node and Workflow objects into the Flyte IDL (Interface Definition Language). The flytekit.tools.translator.get_serializable function converts a WorkflowBase instance into a WorkflowSpec.

This spec contains:

1. Nodes: A list of node templates including their task_node or workflow_node references.
2. Output Bindings: Mapping of workflow outputs to specific node outputs.
3. Upstream Node IDs: The explicit dependency list for each node, ensuring the Flyte engine executes them in the correct order.