Skip to main content

Configuring Task Behavior with Metadata

Configuring task behavior in flytekit is primarily done through the @task decorator, which populates a TaskMetadata object. This metadata controls execution aspects such as retries, timeouts, caching, and resource allocation.

Basic Configuration: Retries and Timeouts​

You can configure how many times a task should be retried upon failure and the maximum duration it is allowed to run.

from datetime import timedelta
from flytekit import task

@task(retries=3, timeout=timedelta(minutes=10))
def compute_heavy_task(x: int) -> int:
    return x * x
  • retries: An integer specifying the number of times to retry the task. A value of n > 0 means the task will be retried at least n times on failure.
  • timeout: Can be a datetime.timedelta or an integer representing seconds. If the task execution exceeds this duration, it will be terminated.

Caching Task Results​

Caching allows flytekit to skip task execution if the same inputs are provided again. To enable caching, you must provide a version string.

Simple Caching​

The simplest way to enable caching is by setting cache=True and providing a cache_version.

@task(cache=True, cache_version="1.0")
def cached_task(x: int) -> int:
    return x + 1

Advanced Caching with the Cache Object​

For more granular control, use the Cache object from flytekit.core.cache. This is the recommended approach for complex caching logic, such as ignoring specific input variables.

from flytekit import task
from flytekit.core.cache import Cache

@task(
    cache=Cache(
        version="v2",
        serialize=True,
        ignored_inputs=("unused_param",)
    )
)
def advanced_cached_task(x: int, unused_param: int) -> int:
    return x * 2
  • version: The version of the task logic. Changing this invalidates the cache.
  • serialize: If True, concurrent executions with the same inputs will run serially. Only one instance executes while others wait to reuse the result, preventing "thundering herd" issues.
  • ignored_inputs: A tuple of input names that should not be included when calculating the cache hash.

Execution Environment and Resources​

You can specify the environment variables and compute resources required for a task.

from flytekit import task, Resources

@task(
    requests=Resources(cpu="2", mem="1Gi"),
    limits=Resources(cpu="4", mem="2Gi"),
    environment={"STAGE": "production"}
)
def resource_intensive_task(data: list) -> int:
    import os
    print(f"Running in {os.environ.get('STAGE')}")
    return len(data)
  • requests: The minimum resources required for the task.
  • limits: The maximum resources the task is allowed to consume.
  • environment: A dictionary of environment variables to be set during execution.

Advanced Metadata Settings​

Interruptible Tasks​

Interruptible tasks can be scheduled on nodes with lower Quality of Service (QoS) guarantees, such as AWS Spot Instances or GCP Preemptible VMs.

@task(interruptible=True)
def spot_instance_task(x: int) -> int:
    return x + 1

Using Pod Templates​

If your task requires specific Kubernetes Pod configurations (like sidecars or special volumes), you can reference a pre-existing PodTemplate.

@task(pod_template_name="my-custom-pod-template")
def pod_template_task(x: int) -> int:
    return x + 1

Deprecation Warnings​

You can mark a task as deprecated by providing a message. This message will be displayed to users when the task is used.

@task(deprecated="This task is deprecated, use 'new_task' instead.")
def old_task(x: int) -> int:
    return x

Troubleshooting​

Caching Errors​

If you enable caching (cache=True) but do not provide a cache_version, flytekit will raise a ValueError during task definition: ValueError: Caching is enabled ``cache=True`` but ``cache_version`` is not set.

Deprecated Arguments​

The arguments cache_serialize, cache_version, and cache_ignore_input_vars should not be passed directly to the @task decorator if you are also using a Cache object. Doing so will result in a ValueError: ValueError: cache_serialize, cache_version, and cache_ignore_input_vars are deprecated. Please use Cache object

Timeout Format​

The timeout parameter must be either an int (seconds) or a datetime.timedelta. Providing other types will raise a ValueError during initialization.