Tasks

The fundamental building block of Flyte — define, configure, and execute Python function tasks.

A task is the fundamental building block and an extension point within Flyte. Tasks have the following characteristics:

• Versioned — typically aligned with the git SHA
• Strongly typed — all inputs and outputs must be annotated
• Declarative — describe what the task does, not how to schedule it
• Independently executable — a task can be run on its own without a workflow
• Unit-testable — call task functions directly in tests

A Flyte task runs in its own container on a Kubernetes pod. This page focuses on Python function tasks — the most common task type.

Basic task

Apply the @task decorator to any Python function. All inputs and outputs must have type annotations:

task.py

from flytekit import task


@task
def slope(x: list[int], y: list[int]) -> float:
    n = len(x)
    sum_xy = sum(xi * yi for xi, yi in zip(x, y))
    sum_x = sum(x)
    sum_y = sum(y)
    sum_x2 = sum(xi ** 2 for xi in x)
    return (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x ** 2)

You can execute a Flyte task like any regular Python function:

result = slope(x=[-3, 0, 3], y=[7, 4, -2])
print(result)  # -1.5

When calling a task function, always use keyword arguments. Positional arguments are not supported.

Run a task with pyflyte

Local

pyflyte run task.py slope --x '[-3,0,3]' --y '[7,4,-2]'

Remote

pyflyte run --remote task.py slope --x '[-3,0,3]' --y '[7,4,-2]'

Multiple outputs

Return a tuple when a task produces multiple outputs. Flyte assigns default names o0, o1, o2, … in positional order:

from flytekit import task
import typing


@task
def stats(values: list[float]) -> typing.Tuple[float, float]:
    mean = sum(values) / len(values)
    variance = sum((v - mean) ** 2 for v in values) / len(values)
    return mean, variance

For human-readable output names, see Named outputs.

Resource requests

Use Resources to request CPU, memory, or GPU for a task:

from flytekit import task, Resources


@task(requests=Resources(cpu="2", mem="2Gi"))
def memory_intensive_task(data: list[float]) -> float:
    return sum(data)


@task(requests=Resources(cpu="4", mem="8Gi", gpu="1"))
def gpu_training_task(epochs: int) -> float:
    # GPU-accelerated training logic
    return 0.95

requests describes the minimum resources Kubernetes should reserve. Use limits to cap the maximum. Both accept the same Resources fields.

Caching

Enable caching to skip re-execution when inputs have not changed. Set cache=True and provide a cache_version string:

from flytekit import task


@task(cache=True, cache_version="1.0")
def expensive_computation(n: int) -> int:
    total = 0
    for i in range(n):
        total += i
    return total

Increment cache_version whenever the task’s logic changes and you want to invalidate old cached results.

Retries

Configure automatic retries for tasks that may fail transiently (e.g., network calls):

from flytekit import task


@task(retries=3)
def fetch_data(url: str) -> str:
    import urllib.request
    with urllib.request.urlopen(url) as response:
        return response.read().decode()

Flyte retries the task up to the specified number of times before marking the execution as failed.

Timeouts

Set a maximum execution time for a task using timeout:

from datetime import timedelta
from flytekit import task


@task(timeout=timedelta(hours=2))
def long_running_task(data: list[float]) -> float:
    import time
    time.sleep(1)
    return sum(data)

Environment variables

Pass environment variables into the task container using environment:

from flytekit import task


@task(environment={"MODEL_ENV": "production", "LOG_LEVEL": "info"})
def configured_task(x: int) -> int:
    import os
    env = os.environ.get("MODEL_ENV", "development")
    print(f"Running in {env}")
    return x * 2

Combining task options

All task options can be combined on a single decorator:

from datetime import timedelta
from flytekit import task, Resources


@task(
    cache=True,
    cache_version="2.1",
    retries=2,
    timeout=timedelta(minutes=30),
    requests=Resources(cpu="2", mem="4Gi"),
    environment={"BATCH_SIZE": "256"},
)
def production_task(dataset: str) -> float:
    # Task logic here
    return 0.0

Task types

Python function task

The most common type. Apply @task to any Python function. Runs the function body in a container on Kubernetes.

from flytekit import task

@task
def my_task(x: int) -> int:
    return x + 1

Shell task

Run bash scripts using ShellTask. See Shell tasks for a full guide.

from flytekit.extras.tasks.shell import ShellTask

t = ShellTask(
    name="my_shell_task",
    script="echo hello world",
)

Plugin tasks

Flyte supports backend plugins for services like AWS Athena, Google BigQuery, SageMaker, Spark, and Ray. These extend the @task interface without requiring custom container code.

Next steps

Workflows

Connect tasks together into a directed acyclic graph.

Named outputs

Assign meaningful names to task outputs using NamedTuple.