Task authoring and execution

Flyte tasks are the fundamental building blocks of execution in flytekit. They represent a single unit of work, characterized by a versioned, strongly-typed interface and independent executability. In flytekit, tasks are primarily authored by decorating Python functions, which the SDK then transforms into a PythonFunctionTask.

Declaring Tasks with @task

The most common way to define a task is using the @task decorator. This decorator inspects your Python function's type hints to generate a Flyte interface and wraps the logic for execution on the Flyte platform.

from flytekit import task
import typing

@task(cache=True, cache_version="1.0", retries=3)
def add_numbers(x: int, y: int) -> int:
    return x + y

When you apply @task, flytekit creates an instance of PythonFunctionTask (defined in python_function_task.py). This class inherits from PythonTask and Task (defined in base_task.py), which provide the core abstractions for task metadata and execution logic.

Task Configuration

The @task decorator accepts several parameters to control how the task behaves on the cluster:

• Caching: Use cache=True and cache_version to avoid re-running tasks with the same inputs. Internally, TaskMetadata (in base_task.py) manages these settings.
• Retries: The retries parameter (an integer) defines how many times Flyte should attempt to re-run the task upon failure.
• Resources: You can specify requests and limits for CPU, memory, and storage.
• Container Image: Use container_image to specify a custom Docker image for the task, overriding the default FLYTE_INTERNAL_IMAGE.
• Timeout: The timeout parameter (either datetime.timedelta or an integer representing seconds) limits the maximum duration of a single execution.

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 locally, 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 enabled.
3. Calls dispatch_execute, which eventually runs your original Python function.
4. Wraps the results back into Promise objects or native Python types.

Remote Execution

On a Flyte cluster, the entry point is typically dispatch_execute. For PythonFunctionTask, this involves:

1. Pre-execution: pre_execute is called to set up the environment (e.g., configuring ExecutionParameters).
2. Input Translation: _literal_map_to_python_input converts the LiteralMap received from the Flyte engine into Python-native keyword arguments.
3. User Code Execution: The execute method runs the decorated function with the translated inputs.
4. Output Translation: _output_to_literal_map converts the function's return values back into a LiteralMap to be sent back to the Flyte engine.

Dynamic Workflows

A dynamic workflow is a special type of task that generates a workflow at runtime based on its inputs. You declare it using the @dynamic decorator, which is a partial application of @task with execution_mode=PythonFunctionTask.ExecutionBehavior.DYNAMIC.

from flytekit import dynamic

@dynamic
def my_dynamic_task(n: int) -> typing.List[int]:
    results = []
    for i in range(n):
        results.append(some_other_task(x=i))
    return results

How Dynamic Tasks Work

Unlike a standard task, a dynamic task's execute method calls dynamic_execute.

• Compilation: During execution on the cluster, the task body is run. Instead of just returning values, it "records" the tasks called inside it.
• Workflow Generation: compile_into_workflow (in python_function_task.py) produces a DynamicJobSpec. This spec contains a complete workflow template (nodes, outputs, and subworkflows) generated on the fly.
• Backend Hand-off: The Flyte backend receives this spec and executes it as a subworkflow.

Core Abstractions

The task system is built on a hierarchy of classes in base_task.py and python_function_task.py:

Class	Purpose
Task	The base abstraction matching the Flyte IDL TaskTemplate. It handles metadata like retries and caching.
PythonTask	Adds a python_interface to the base Task, enabling mapping between Python types and Flyte types.
PythonFunctionTask	The implementation used for @task. It holds a reference to the actual Python task_function.
TaskMetadata	A container for execution-related settings like timeout, interruptible, and cache_version.

Task Resolvers

When a task runs on a cluster, Flyte needs to know how to find and load the Python code. This is handled by TaskResolverMixin. The default_task_resolver identifies a task by its module and name. When pyflyte-execute runs in a container, it uses the resolver to call load_task, which imports the module and retrieves the task object.