Task authoring and execution

Flyte tasks are the fundamental building blocks of a Flyte workflow. They represent a single unit of work, characterized by a strong interface (typed inputs and outputs), versioning, and independent executability. In flytekit, tasks are primarily declared using the @task decorator, which transforms a standard Python function into a PythonFunctionTask.

Declaring Tasks

You define a task by decorating a Python function with @task. flytekit uses the function's type hints to automatically generate a TypedInterface that Flyte uses for data validation and orchestration.

import typing
from flytekit import task

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

@task
def process_data(x: int, y: typing.Dict[str, str]) -> str:
    # flytekit automatically detects inputs and outputs from type hints
    return f"Processed {x} with {len(y)} entries"

When you call this function, flytekit intercepts the call via flyte_entity_call_handler. In a local environment, it executes the function directly. In a workflow context, it creates a Promise representing the future output of the task.

Task Configuration and Metadata

The @task decorator accepts several parameters to control execution behavior, resource allocation, and caching. These settings are encapsulated in the TaskMetadata class internally.

Caching and Retries

To avoid redundant computations, you can enable caching. If a task is called with the same inputs and the same cache_version, Flyte will return the cached result instead of re-executing.

from flytekit import task, Cache

@task(
    cache=True, 
    cache_version="1.0", 
    retries=3, 
    timeout=3600
)
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.

Resource Requests and Limits

You can specify the compute resources required for a task using the requests and limits parameters with the Resources class.

from flytekit import task, Resources

@task(
    requests=Resources(cpu="1", mem="2Gi"),
    limits=Resources(cpu="2", mem="4Gi"),
    environment={"STAGE": "production"}
)
def memory_intensive_task(n: int) -> list[int]:
    return [i for i in range(n)]

Core Task Abstractions

flytekit provides a hierarchy of classes to handle different task types:

1. Task: The base class in flytekit.core.base_task. it maps closely to the Flyte IDL TaskTemplate. It handles the core logic for local_execute and dispatch_execute.
2. PythonTask: A base class for tasks that have a Python-native Interface. It manages the translation between Flyte literals and Python native types using the TypeEngine.
3. PythonFunctionTask: The most common task type, used by the @task decorator. It wraps a user-defined Python function and handles its execution.

Task Execution Flow

When a task is executed (either locally or on a cluster), the following sequence occurs within PythonFunctionTask.dispatch_execute:

1. pre_execute: Prepares the execution environment (e.g., setting up a Spark session).
2. Input Translation: Converts LiteralMap (Flyte's internal data format) into Python native types using _literal_map_to_python_input.
3. execute: Invokes the actual Python function with the translated inputs.
4. post_execute: Allows for cleanup or output modification.
5. Output Translation: Converts the Python return values back into a LiteralMap via _output_to_literal_map.

Specialized Task Types

Async and Eager Tasks

flytekit supports asynchronous tasks using AsyncPythonFunctionTask. If you decorate an async def function with @task, flytekit automatically selects this task type.

Eager tasks (declared with @eager) allow for dynamic execution logic where the Python code itself acts as the orchestrator, creating stack frames on the Flyte cluster for each task invocation.

from flytekit import eager, task

@task
async def get_data() -> int:
    return 42

@eager
async def eager_workflow(x: int) -> int:
    # This runs on the cluster but allows imperative logic
    data = await get_data()
    if data > x:
        return data
    return x

Reference Tasks

If a task is already registered on a Flyte cluster, you can reference it without redefining the logic using the @reference_task decorator. This is useful for cross-project task sharing.

from flytekit import reference_task

@reference_task(
    project="flytesnacks",
    domain="development",
    name="core.recipes.simple.greet",
    version="v1"
)
def remote_greet(name: str) -> str:
    ...

Task Decks

Flyte Decks provide a way to visualize task inputs, outputs, and custom data in the Flyte UI. By default, PythonFunctionTask generates decks for source code, dependencies, and timeline. You can control this via enable_deck and deck_fields.

from flytekit import task, DeckField

@task(enable_deck=True, deck_fields=(DeckField.INPUT, DeckField.OUTPUT))
def deck_task(x: int) -> int:
    return x + 1

Internally, PythonFunctionTask._write_decks uses TypeEngine.to_html to render the data into the deck.