Workflow composition and nodes

Flyte workflows are the primary mechanism for composing tasks and other workflows into a Directed Acyclic Graph (DAG). In flytekit, you define these graphs using the @workflow decorator on a Python function.

Composing Workflows

When you decorate a function with @workflow, flytekit treats the function body as a declaration of the workflow's structure. Instead of executing the code immediately, flytekit "compiles" the function by tracking calls to tasks and other workflows.

from flytekit import task, workflow

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

@task
def greet(greeting: str, name: str) -> str:
    return f"{greeting} How are you, {name}?"

@workflow
def greeting_workflow(name: str) -> str:
    # Calling a task creates a Node in the graph
    hello_msg = say_hello(name=name)
    # Passing the output of one task to another creates a data dependency
    return greet(greeting=hello_msg, name=name)

The Workflow Body as a Declaration

It is critical to understand that the body of a @workflow function is executed at compile time. The objects returned by tasks (like hello_msg above) are not actual strings or integers; they are Promise objects.

Because of this, you cannot perform standard Python operations on these values inside the workflow body. For example, if hello_msg == "Hello, world!": will not work as expected because hello_msg is a Promise, not a string. All data manipulation must happen inside tasks.

Nodes and Execution Order

Every time you call a task or a sub-workflow within a workflow, flytekit creates a Node. A Node (defined in flytekit.core.node.Node) encapsulates the underlying entity, its inputs, and its position in the graph.

Data Dependencies

The most common way to connect nodes is through data dependencies. When you pass the output of Task A as an input to Task B, flytekit automatically ensures that Task A completes before Task B starts.

Explicit Dependencies with >>

Sometimes you need to enforce an execution order even when there is no data being passed between tasks. You can use the right-shift operator >> to define these dependencies.

@workflow
def ordered_workflow():
    t1_node = t1()
    t2_node = t2()
    
    # Ensure t1 runs before t2
    t1_node >> t2_node

Internally, the Node.__rshift__ method calls Node.runs_before(other), which appends the current node to the _upstream_nodes list of the target node.

Customizing Node Behavior

You can customize the execution parameters of a specific node without changing the underlying task definition by using the with_overrides method. This is useful for setting resource limits, retries, or timeouts for a specific step in a workflow.

@workflow
def resource_workflow(n: int) -> int:
    return t1(n=n).with_overrides(
        requests=Resources(cpu="2", mem="200Mi"),
        limits=Resources(cpu="4", mem="500Mi"),
        retries=3,
        timeout=datetime.timedelta(minutes=10)
    )

The with_overrides method in flytekit.core.node.Node handles several types of customizations:

• Resources: Sets requests and limits using the Resources class.
• Retries: Configures a RetryStrategy in the node's metadata.
• Timeout: Overrides the default task timeout.
• Container Image: Allows specifying a different container image for that specific node.

Note: You cannot use Promise objects (outputs from other tasks) for override values like retries or requests. These must be static values known at compile time.

Manual Node Creation

For advanced scenarios where you need to create nodes dynamically or handle complex dependencies, you can use the create_node function from flytekit.core.node_creation.

from flytekit.core.node_creation import create_node

@workflow
def manual_node_wf(a: int):
    # Manually create a node for task t1
    t1_node = create_node(t1, a=a)
    
    # Access outputs via the node object
    t2(b=t1_node.o0)

When create_node is called during compilation, it invokes the entity (task or workflow) to generate the necessary Promise objects and then retrieves the newly created node from the FlyteContext. The outputs of the node are attached as attributes (e.g., node.o0, node.o1) to allow them to be passed to subsequent tasks.

Limitations

• create_node cannot be used inside conditional blocks.
• In local execution, create_node returns the actual results (or a tuple of results) rather than a Node object, to maintain compatibility with Python's execution model.