Skip to main content

Task authoring and execution

Flyte tasks are the fundamental building blocks of a workflow. In flytekit, a task is a versioned, independently executable entity with a strongly typed interface. While the underlying engine treats tasks as language-agnostic templates, flytekit provides a Pythonic way to define, configure, and execute them using the @task decorator.

Declaring Tasks​

The primary way to create a task in flytekit is by decorating a Python function with @task. This decorator transforms a standard Python function into an instance of PythonFunctionTask (defined in flytekit.core.python_function_task).

from flytekit import task
import typing

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

When you decorate a function, flytekit performs several internal actions:

  1. Interface Discovery: It uses transform_function_to_interface (from flytekit.core.interface) to inspect the function's type hints and generate a TypedInterface.
  2. Metadata Creation: It initializes a TaskMetadata object (from flytekit.core.base_task) to store configuration like retries and timeouts.
  3. Plugin Selection: It uses TaskPlugins.find_pythontask_plugin to determine the correct task class. By default, this is PythonFunctionTask, but it can be overridden by providing a task_config.

Task Configuration​

The @task decorator accepts various arguments to control how the task is executed on the Flyte cluster. These settings are captured in the TaskMetadata class and the PythonTask base class.

Retries and Timeouts​

You can protect against transient failures or runaway processes by setting retries and timeouts.

from datetime import timedelta

@task(retries=3, timeout=timedelta(minutes=5))
def flaky_task(x: int) -> int:
    ...

Internally, TaskMetadata validates these values. For example, it ensures timeout is either an integer (seconds) or a timedelta object.

Resource Management​

Tasks can request specific compute resources like CPU, memory, and storage using the Resources class.

from flytekit import Resources

@task(
    requests=Resources(cpu="2", mem="500Mi"),
    limits=Resources(cpu="4", mem="1Gi")
)
def resource_intensive_task(data: list) -> int:
    ...

Caching​

Flyte can cache task results to avoid redundant computations. Caching is enabled by passing cache=True and a cache_version.

@task(cache=True, cache_version="1.0")
def expensive_computation(a: int, b: int) -> int:
    ...

The TaskMetadata class requires cache_version if cache is enabled. During local execution, Task.local_execute checks the LocalTaskCache before running the task body.

Execution Behaviors​

Flytekit supports different execution modes through the PythonFunctionTask.ExecutionBehavior enum.

Standard Execution​

By default, a task runs as a single containerized unit. The dispatch_execute method in PythonFunctionTask handles the translation of Flyte LiteralMap inputs into Python native values, calls the user function, and translates the results back into Flyte literals.

Dynamic Tasks​

A dynamic task is declared using the @dynamic decorator (which is a partial application of @task with execution_mode=ExecutionBehavior.DYNAMIC). Unlike a standard task, a dynamic task's body is executed to generate a new workflow at runtime.

from flytekit import dynamic

@dynamic
def dynamic_subwf(n: int) -> typing.List[int]:
    results = []
    for i in range(n):
        results.append(greet(name=str(i)))
    return results

When PythonFunctionTask.execute is called in dynamic mode, it invokes dynamic_execute. In a remote environment, this triggers compile_into_workflow, which produces a DynamicJobSpec containing the generated nodes and tasks.

Eager Tasks​

Eager tasks (using @task(execution_mode=ExecutionBehavior.EAGER)) allow for more flexible, imperative-style execution where Python code acts as the orchestrator. This is implemented by EagerAsyncPythonFunctionTask. When running on the backend, it uses a Controller to manage sub-executions and can even render a "Timeline Deck" to visualize the execution flow.

Internal Task Hierarchy​

Flytekit uses a class hierarchy to separate the abstract definition of a task from its Python-specific implementation:

  1. Task (flytekit.core.base_task): The base class closest to the Flyte IDL. It defines the core interface for dispatch_execute, pre_execute, and post_execute.
  2. PythonTask (flytekit.core.base_task): Adds support for Python-native interfaces and handles the conversion between Python types and Flyte literals via the TypeEngine.
  3. PythonFunctionTask (flytekit.core.python_function_task): The concrete implementation for tasks defined as Python functions. It stores the task_function and manages the different ExecutionBehavior modes.

Task Resolvers​

When a task runs in a container on the Flyte cluster, the engine needs to know how to find and load the Python code. This is handled by TaskResolverMixin. The default_task_resolver (in flytekit.core.python_auto_container) identifies tasks by their module and name.

If you use custom decorators, ensure they use functools.wraps, as PythonFunctionTask validates that the function is accessible at the module level to ensure the resolver can rehydrate the task object during remote execution.