Defining Tasks and Workflows

Tasks and workflows are the fundamental building blocks of flytekit. By using Python decorators, you can transform standard Python functions into robust, versioned, and independently executable units of work that Flyte can orchestrate.

In this tutorial, you will build a simple text processing pipeline that greets a user and counts the length of the greeting.

Prerequisites

To follow this tutorial, you must have flytekit installed:

pip install flytekit

Step 1: Defining Tasks

A task is the smallest unit of execution in flytekit. You define a task by decorating a Python function with the @task decorator. This transforms your function into an instance of PythonFunctionTask (found in flytekit.core.python_function_task).

Flytekit uses Python type hints to define the task's interface. These hints are mandatory because flytekit uses them to ensure type safety across the entire pipeline.

from flytekit import task

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

@task
def count_letters(text: str) -> int:
    return len(text)

When you call get_greeting, flytekit's PythonFunctionTask uses transform_function_to_interface to inspect the name: str input and -> str output. This information is captured in a TypedInterface which Flyte uses to validate data flow.

Step 2: Defining Workflows

A workflow composes multiple tasks into a directed acyclic graph (DAG). You define a workflow by decorating a function with the @workflow decorator, which creates a PythonFunctionWorkflow (found in flytekit.core.workflow).

Unlike tasks, the body of a workflow function is not used for data processing. Instead, flytekit "traces" the function to understand how tasks are connected.

from flytekit import workflow

@workflow
def greeting_workflow(name: str) -> int:
    greeting = get_greeting(name=name)
    count = count_letters(text=greeting)
    return count

In the example above, get_greeting(name=name) does not return a string during the tracing phase. Instead, it returns a Promise object (from flytekit.core.promise). This Promise is passed as an input to count_letters, creating a dependency between the two tasks.

Step 3: Adding Configuration

You can configure task behavior such as retries and caching directly in the @task decorator. These configurations are stored in TaskMetadata within the Task base class (found in flytekit.core.base_task).

Caching

Caching allows you to skip task execution if the inputs and the cache_version remain the same.

@task(cache=True, cache_version="1.0")
def expensive_computation(x: int) -> int:
    # This result will be cached
    return x * x

[!IMPORTANT] When cache=True is set, you must provide a cache_version. If the logic inside your task changes, you should manually bump this version to invalidate the old cache.

Retries

You can specify how many times Flyte should retry a task in case of failure:

@task(retries=3)
def unstable_task(x: int) -> int:
    ...

Step 4: Local Execution

One of the key features of flytekit is that you can run your tasks and workflows locally just like standard Python functions. This is handled by the local_execute methods in Task and WorkflowBase.

You can run your workflow by calling it at the end of your script:

if __name__ == "__main__":
    # This executes the workflow locally
    result = greeting_workflow(name="Flyte")
    print(f"The greeting length is: {result}")

When you run this script, flytekit enters LOCAL_WORKFLOW_EXECUTION mode. It translates your Python native inputs into Flyte literals, executes the tasks in the correct order, and returns the final result.

Complete Example

Here is the complete code for your first flytekit project:

from flytekit import task, workflow

@task(cache=True, cache_version="1.0")
def get_greeting(name: str) -> str:
    return f"Hello, {name}!"

@task(retries=3)
def count_letters(text: str) -> int:
    return len(text)

@workflow
def greeting_workflow(name: str) -> int:
    greeting = get_greeting(name=name)
    count = count_letters(text=greeting)
    return count

if __name__ == "__main__":
    print(f"Result: {greeting_workflow(name='World')}")

By completing this tutorial, you have learned how to use PythonFunctionTask and PythonFunctionWorkflow to build a type-safe, configurable pipeline. Next, you can explore Dynamic Tasks for workflows where the graph structure depends on runtime data.