Skip to main content

Task authoring and execution

Flyte tasks are the fundamental building blocks of a Flyte workflow. They represent a discrete unit of work, encapsulated with a strong interface, versioning, and specific execution requirements. In flytekit, tasks are typically defined by decorating a Python function, which the framework then transforms into a structured task object capable of running locally or on a remote Flyte cluster.

Declaring Tasks​

The primary way to define a task in flytekit is using the @task decorator found in flytekit/core/task.py. This decorator transforms a standard Python function into a PythonFunctionTask.

from flytekit import task
import typing

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

When you apply @task, flytekit uses transform_function_to_interface to inspect the function's type hints and docstrings. This creates a TypedInterface that Flyte uses to validate data flow between tasks.

Task Metadata and Configuration​

The @task decorator accepts several parameters to control execution behavior, which are stored in the TaskMetadata class (defined in flytekit/core/base_task.py).

  • Retries: Use retries to specify how many times Flyte should attempt to re-run the task on failure.
  • Caching: Enable cache=True and provide a cache_version to avoid redundant computations.
  • Timeouts: Use timeout (as an integer or datetime.timedelta) to limit the maximum execution time.
  • Resources: Specify requests and limits using the Resources class to request specific CPU, memory, or GPU allocations.
from datetime import timedelta
from flytekit import task, Resources

@task(
    retries=3,
    cache=True,
    cache_version="1.0",
    timeout=timedelta(minutes=5),
    requests=Resources(cpu="2", mem="500Mi"),
    limits=Resources(cpu="4", mem="1Gi"),
)
def resource_intensive_task(x: int) -> int:
    return x * x

Core Task Abstractions​

Flytekit uses a hierarchy of classes to manage different task types:

  1. Task: The base class in flytekit/core/base_task.py. it captures the FlyteIDL TaskTemplate information but lacks a Python-native interface.
  2. PythonTask: A subclass of Task that introduces a python_interface. It handles the translation between Flyte's internal Literal types and Python native types.
  3. PythonFunctionTask: The most common implementation, which wraps a user-defined Python function. It manages the execution logic via its execute method.

The Execution Lifecycle​

When a task is executed, flytekit follows a structured dispatch process implemented in PythonTask.dispatch_execute:

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

Specialized Task Behaviors​

Dynamic Tasks​

A task can be marked as dynamic by using the @dynamic decorator (which sets execution_mode to ExecutionBehavior.DYNAMIC). Dynamic tasks allow you to generate new workflow nodes at runtime based on the task's inputs. Internally, PythonFunctionTask.compile_into_workflow is called during execution to produce a DynamicJobSpec.

Eager Tasks​

Eager tasks (implemented via EagerAsyncPythonFunctionTask) allow for a more imperative, "Pythonic" style of workflow where tasks are awaited. Unlike standard tasks that return Promise objects during workflow compilation, eager tasks can execute and return actual values by interacting with a Controller that manages sub-executions on the Flyte backend.

Async Tasks​

If you decorate an async def function with @task, flytekit instantiates an AsyncPythonFunctionTask. These tasks are executed within an asynchronous loop managed by flytekit's loop_manager.

Task Resolvers​

When a task runs on a remote cluster, the container needs to know how to find and load the specific Python function. This is handled by a TaskResolverMixin. The default_task_resolver in flytekit/core/python_auto_container.py works by:

  1. Capturing the module and function name during serialization.
  2. Generating loader_args (e.g., task-module my_module task-name my_task).
  3. Using importlib to re-import the task in the execution container.

Constraints and Requirements​

  • Module Level: Tasks must be defined at the module level so they can be imported by the task resolver. Nested or local functions will raise a ValueError unless they are in a test module (files starting with test_).
  • Type Hints: Every input and output must have a type hint. Flytekit uses these to build the TypedInterface.
  • Decks: Flyte Decks provide visibility into task execution. While disable_deck is supported for backward compatibility, you should use enable_deck=True to generate HTML reports for inputs, outputs, and system metadata.