Task authoring and execution

Flyte tasks are the fundamental building blocks of a workflow. In flytekit, a task is a versioned, independently executable unit of logic with a strongly typed interface. While the underlying system uses the Task base class to map to Flyte IDL specifications, most developers interact with tasks through the @task decorator and the PythonFunctionTask class.

Declaring Tasks

You define a task by decorating a Python function with @task. flytekit uses the function's type hints to automatically generate the task's interface, including input and output names and types.

from flytekit import task
import typing

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

@task
def add_numbers(a: int, b: int) -> int:
    return a + b

When you decorate a function, flytekit creates an instance of PythonFunctionTask (defined in python_function_task.py). This class captures the function body, the detected interface, and any configuration provided to the decorator.

Task Metadata and Configuration

The @task decorator accepts various parameters to control execution behavior, which are stored in the TaskMetadata class (found in base_task.py). Common configurations include:

• Retries: The number of times to retry the task on failure.
• Caching: Enable caching by setting cache=True and providing a cache_version.
• Timeout: A datetime.timedelta or integer (seconds) representing the maximum duration for a single execution.
• Interruptible: A boolean indicating if the task can run on lower-cost, pre-emptible nodes.

from flytekit import task
from datetime import timedelta

@task(
    retries=3,
    cache=True,
    cache_version="1.0",
    timeout=timedelta(minutes=5),
    interruptible=True
)
def heavy_computation(data: list[int]) -> int:
    return sum(data)

Internally, TaskMetadata validates these settings. For example, it ensures that if cache=True, a cache_version is also provided, and that cache_serialize is only enabled if caching is active.

Task Execution Flow

Flytekit handles task execution differently depending on whether it is running locally or on a remote Flyte cluster.

Local Execution

When you call a task function directly in a Python script, flytekit invokes Task.local_execute. This method:

1. Translates Python native inputs into Flyte literals using translate_inputs_to_literals.
2. Checks the local cache if caching is enabled via LocalTaskCache.get.
3. Executes the task logic.
4. Wraps the results back into Promise objects or native Python types.

Dispatch and Sandbox Execution

During remote execution or local sandbox testing, the system calls dispatch_execute. This method is responsible for the full lifecycle of a task run:

1. Pre-execution: pre_execute is called to set up the environment (e.g., initializing a Spark session).
2. Input Translation: Input literals are converted to Python native types using _literal_map_to_python_input.
3. User Code Execution: The actual execute method (which wraps your decorated function) is invoked.
4. Post-execution: post_execute allows for cleanup or output modification.
5. Output Translation: Results are converted back to a LiteralMap via _output_to_literal_map.

Specialized Task Types

Flytekit provides specialized task abstractions for complex execution patterns.

Dynamic Tasks

A dynamic task, declared with @dynamic, is a task that generates a workflow at runtime based on its inputs. This is implemented by PythonFunctionTask with an execution_mode of ExecutionBehavior.DYNAMIC.

from flytekit import task, dynamic

@task
def process_item(item: int) -> int:
    return item * 2

@dynamic
def process_list(items: list[int]) -> list[int]:
    return [process_item(item=i) for i in items]

When a dynamic task executes, it calls compile_into_workflow, which produces a DynamicJobSpec containing the generated nodes and task templates.

Map Tasks

Map tasks allow you to run a single task across a list of inputs in parallel. You use map_task to wrap an existing task.

from flytekit import task, workflow, map_task

@task
def square(val: int) -> int:
    return val * val

@workflow
def my_map_workflow(inputs: list[int]) -> list[int]:
    return map_task(square)(val=inputs)

Eager Tasks

Eager tasks (or eager workflows), declared with @eager, allow for fully dynamic Pythonic execution where each task call results in a separate execution on the Flyte cluster. These are implemented by EagerAsyncPythonFunctionTask. Unlike dynamic tasks, eager tasks are not compiled into a static spec but use asyncio to manage remote executions via a Controller.

Task Resolvers

When a task runs in a container on a Flyte cluster, the system needs to know how to find and load the original Python task object. This is handled by TaskResolverMixin. The default_task_resolver captures the module and function name of the task. At runtime, pyflyte-execute uses these arguments to import the module and retrieve the task instance for execution.