Task authoring and execution

Flyte tasks are the fundamental building blocks of a workflow. In flytekit, tasks are defined by decorating a Python function with the @task decorator, which transforms the function into a PythonFunctionTask. This abstraction handles the translation between Python native types and the Flyte IDL, manages execution metadata, and provides hooks for plugin-specific configurations.

Defining Tasks with @task

The @task decorator is the primary entry point for authoring tasks. It automatically infers the task's interface (inputs and outputs) from Python type hints using transform_function_to_interface in flytekit/core/interface.py.

from flytekit import task

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

When you call this function, flytekit intercepts the call via flyte_entity_call_handler. Depending on the ExecutionState, it either executes the function locally or compiles it into a node for a workflow.

Task Metadata and Configuration

You can configure execution behavior such as retries, timeouts, and caching by passing arguments to the @task decorator. These are stored in the TaskMetadata class (found in flytekit/core/base_task.py).

from datetime import timedelta
from flytekit import task, Cache

@task(
    retries=3,
    timeout=timedelta(minutes=60),
    cache=True,
    cache_version="1.0",
    interruptible=True
)
def compute_expensive_result(data: list[int]) -> int:
    return sum(data)

Key metadata attributes include:

• retries: The number of times the task is retried on failure.
• timeout: The maximum duration for a single execution.
• cache and cache_version: Enables memoization. If cache=True, you must provide a cache_version.
• interruptible: Indicates the task can run on lower-priority (e.g., spot) instances.

Resource Requests and Limits

Tasks can specify the compute resources they require using the Resources class.

from flytekit import task, Resources

@task(requests=Resources(cpu="2", mem="500Mi"), limits=Resources(cpu="4", mem="1Gi"))
def memory_intensive_task(size: int) -> list[int]:
    return [i for i in range(size)]

Task Execution Flow

The execution of a task follows a structured lifecycle managed by the Task and PythonTask base classes in flytekit/core/base_task.py.

1. Local Execution

When a task is called outside of a workflow context, local_execute is triggered.

1. Input Translation: translate_inputs_to_literals converts Python native values into Flyte Literal objects.
2. Cache Check: If caching is enabled, flytekit checks LocalTaskCache for a hit.
3. Sandbox Execution: sandbox_execute prepares the execution environment and calls dispatch_execute.
4. Output Translation: Results are converted back from Literal objects to Python types.

2. Dispatch Execution

The dispatch_execute method is the core execution engine used both locally and at runtime on the Flyte platform.

• pre_execute: Sets up the environment (e.g., initializing a Spark session).
• execute: Invokes the actual user-defined Python function.
• post_execute: Performs cleanup or output modification.

Specialized Task Types

Flytekit provides specialized task behaviors through the execution_mode parameter in PythonFunctionTask.

Dynamic Tasks

Dynamic tasks allow you to generate a workflow structure at runtime based on inputs. You define them using the @dynamic decorator, which sets the execution mode to DYNAMIC.

from flytekit import dynamic, task

@task
def process_item(item: int) -> int:
    return item * 2

@dynamic
def process_list(items: list[int]) -> list[int]:
    return [process_item(item=i) for i in items]

Internally, dynamic_execute compiles the user code into a DynamicJobSpec at runtime, which Flyte Propeller then executes as a sub-workflow.

Eager Tasks

Eager tasks (defined via EagerAsyncPythonFunctionTask) allow for asynchronous, imperative-style execution where Python code acts as the orchestrator. Every task call within an eager task creates a new execution on the Flyte cluster rather than just a local function call.

Extending Tasks with Plugins

The task_config parameter allows you to integrate with external systems like Spark, SQLAlchemy, or Kubernetes Pods. Flytekit uses a TaskPlugins factory to map configuration types to specific PythonFunctionTask implementations.

from flytekit import task
from flytekitplugins.spark import Spark

@task(
    task_config=Spark(
        spark_conf={"spark.driver.memory": "2g"},
        hadoop_conf={"fs.s3a.access.key": "my-key"}
    )
)
def spark_task(data: list[int]) -> int:
    # This function will be executed within a Spark cluster context
    ...

When task is called with a task_config, TaskPlugins.find_pythontask_plugin(type(task_config)) resolves the appropriate plugin class to instantiate.