Getting Started with Tasks and Workflows

Learn the fundamental building blocks of flytekit by defining your first tasks and workflows. This guide walks you through creating functional and imperative workflows, configuring task metadata, and setting up launch plans for scheduling.

Prerequisites

To follow this tutorial, ensure you have flytekit installed:

pip install flytekit

Defining Tasks with Metadata

Tasks are the basic units of execution in flytekit. You define them using the @task decorator, which transforms a regular Python function into a flytekit.core.base_task.PythonTask. You can configure execution behavior like retries and caching using the flytekit.core.base_task.TaskMetadata attributes passed directly to the decorator.

import datetime
from flytekit import task

@task(
    retries=3,
    cache=True,
    cache_version="1.0",
    timeout=datetime.timedelta(minutes=5)
)
def square(x: int) -> int:
    """
    A simple task that squares an integer.
    """
    return x * x

@task
def sum_values(x: int, y: int) -> int:
    return x + y

In this example:

• retries=3: If the task fails, flytekit will retry it up to 3 times.
• cache=True and cache_version="1.0": Enables caching. If the task is called with the same input, flytekit will return the cached result instead of re-executing.
• timeout: Limits the execution time to 5 minutes.

Composing Workflows

Workflows compose multiple tasks into a directed acyclic graph (DAG). You define them using the @workflow decorator, which creates a flytekit.core.workflow.PythonFunctionWorkflow.

from flytekit import workflow

@workflow
def math_workflow(x: int, y: int) -> int:
    """
    A workflow that squares two numbers and returns their sum.
    """
    val1 = square(x=x)
    val2 = square(x=y)
    return sum_values(x=val1, y=val2)

You can execute this workflow locally just like a standard Python function:

if __name__ == "__main__":
    result = math_workflow(x=2, y=3)
    print(f"Result: {result}")  # Output: Result: 13

Configuring Workflow Failure Policies

You can control how a workflow behaves when a node fails using flytekit.core.workflow.WorkflowFailurePolicy. By default, workflows fail immediately, but you can configure them to continue executing other independent nodes.

from flytekit import workflow
from flytekit.core.workflow import WorkflowFailurePolicy

@workflow(on_failure=WorkflowFailurePolicy.FAIL_AFTER_EXECUTABLE_NODES_COMPLETE)
def robust_workflow(x: int) -> int:
    return square(x=x)

Creating Launch Plans

A flytekit.core.launch_plan.LaunchPlan allows you to define default or fixed inputs for a workflow and set up schedules or notifications.

from flytekit import LaunchPlan
from flytekit.models.schedule import Schedule
from flytekit.core.notification import Email
from flytekit.models.core.execution import WorkflowExecutionPhase

# Define a schedule that runs every minute
minute_schedule = Schedule(
    kickoff_time_input_arg="kickoff_time",
    rate=Schedule.FixedRate(value=1, unit=Schedule.FixedRateUnit.MINUTE)
)

# Define a notification for successful executions
success_notification = Email(
    phases=[WorkflowExecutionPhase.SUCCEEDED],
    recipients_email=["dev-alerts@example.com"]
)

# Create the launch plan
math_lp = LaunchPlan.get_or_create(
    name="math_lp_scheduled",
    workflow=math_workflow,
    default_inputs={"x": 10, "y": 5},
    schedule=minute_schedule,
    notifications=[success_notification]
)

Programmatic Workflow Construction

While the @workflow decorator is the most common way to define workflows, you can also build them programmatically using flytekit.core.workflow.ImperativeWorkflow. This is useful when the workflow structure depends on dynamic logic.

from flytekit import Workflow

# Create the workflow container
wb = Workflow(name="imperative_math_wf")

# Add workflow inputs
in_x = wb.add_workflow_input("x", int)
in_y = wb.add_workflow_input("y", int)

# Add tasks as nodes and bind their inputs
node_square_x = wb.add_entity(square, x=in_x)
node_square_y = wb.add_entity(square, x=in_y)
node_sum = wb.add_entity(sum_values, x=node_square_x.outputs["o0"], y=node_square_y.outputs["o0"])

# Define the workflow output
wb.add_workflow_output("result", node_sum.outputs["o0"])

# Execute the imperative workflow locally
if __name__ == "__main__":
    res = wb(x=4, y=5)
    print(f"Imperative Result: {res}") # Output: Imperative Result: 41

In an ImperativeWorkflow:

• add_workflow_input defines the top-level interface.
• add_entity adds a task or another workflow as a node.
• node.outputs["o0"] accesses the default output of a task (flytekit names single outputs o0 by default).
• add_workflow_output binds a node's output to the workflow's final result.

Next Steps

Now that you have defined tasks and workflows, you can explore:

• Data Types: Use flytekit.models.interface.Variable to define complex interfaces.
• Documentation: Add rich metadata to your entities using flytekit.models.documentation.Documentation.
• Remote Execution: Use flytekit.remote.remote.FlyteRemote to trigger these workflows on a deployed Flyte cluster.