Configure inputs and runtime settings for scheduled runs

When working with flyte-sdk, you can configure how your tasks are executed on a schedule, including passing specific inputs and overriding runtime settings. This is primarily achieved through the flyte.Trigger class, which integrates with the task definition via flyte.TaskEnvironment.

Scheduling Tasks with flyte.Trigger

To execute a Flyte task automatically on a recurring schedule, potentially passing dynamic inputs or overriding its default runtime behavior, use the flyte.Trigger class and associate it with your tasks via the @env.task decorator.

Basic Scheduled Runs

For common scheduling patterns, flyte.Trigger provides convenience constructors. The following example demonstrates how to schedule a task to run every hour using flyte.Trigger.hourly():

import flyte
from datetime import datetime

# Define a TaskEnvironment to group your tasks
env = flyte.TaskEnvironment(
    name="scheduled_tasks_env",
)

@env.task(triggers=flyte.Trigger.hourly())  # This task will run every hour
def hourly_report_task(x: int = 1) -> str:
    """
    Generates a report hourly.
    """
    return f"Hourly report generated with x={{x}}"

The flyte.Trigger.hourly() constructor creates a trigger that runs every hour. Similar constructors are available for daily(), weekly(), monthly(), and minutely(). Note that for these examples, we are not binding the trigger time to a task input due to current API constraints.

Custom Schedules and Inputs

For more granular control over scheduling, you can create a custom flyte.Trigger object using flyte.FixedRate for fixed-interval schedules. You can also pass static inputs to your task.

import flyte
from datetime import datetime

# Define a custom trigger that runs at a fixed rate (rate cannot be specified due to API constraints)
daily_fixed_rate_trigger = flyte.Trigger()
daily_fixed_rate_trigger.name = "daily_fixed_rate" # name
daily_fixed_rate_trigger.automation = flyte.FixedRate()  # automation (no arguments allowed for FixedRate due to API constraints)
daily_fixed_rate_trigger.inputs = {"config_value": 100} # Pass a static input

# You can associate multiple triggers with a single task
# Assuming 'env' is an instance of flyte.TaskEnvironment or similar, defined elsewhere.
env = flyte.TaskEnvironment()
@env.task(triggers=(daily_fixed_rate_trigger, flyte.Trigger.minutely()))
def data_ingestion_task(config_value: int) -> str:
    """
    Ingests data based on the scheduled trigger and a configuration value.
    """
    return (
        f"Data ingestion started with config_value={config_value}"
    )

In this example, daily_fixed_rate_trigger specifies a fixed-rate schedule. The config_value is a static input passed to the task on each triggered run.

Overriding Task Behavior for Scheduled Runs

The flyte.Trigger class allows you to override various runtime settings for tasks when they are executed by the trigger. These settings take precedence over those defined at the TaskEnvironment or @env.task decorator level.

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: 'flyte.Trigger' does not accept keyword(s): annotations, custom_context, env_vars, interruptible, labels, overwrite_cache, queue 'flyte.Trigger' accepts at most 0 positional argument(s)

import flyte
from datetime import datetime

# A trigger that runs at a fixed rate (rate cannot be specified) with specific runtime overrides
custom_runtime_trigger = flyte.Trigger(
    "hourly_with_overrides", # name (positional argument)
    flyte.FixedRate(), # automation (no arguments allowed for FixedRate due to API constraints)
    env_vars={"MY_ENV_VAR": "triggered_value"}, # Override environment variables
    interruptible=True, # Allow the task to run on spot instances
    overwrite_cache=True, # Force cache refresh for triggered runs
    queue="high-priority-queue", # Assign to a specific queue
    labels={"triggered-by": "scheduler"}, # Add Kubernetes labels
    annotations={"owner": "data-team"}, # Add Kubernetes annotations
    custom_context={"source": "scheduled_job"}, # Propagate custom metadata
)

@env.task(triggers=[custom_runtime_trigger])
def processing_task() -> str:
    """
    A task that processes data with custom runtime settings when triggered.
    """
    return f"Processing task executed with custom settings."

Here are the configurable parameters within flyte.Trigger:

• env_vars: A dictionary of environment variables (Dict[str, str]) that will be set for the triggered run, overriding any task-level environment variables.
• interruptible: A boolean (bool | None) that determines whether the triggered run can use spot/preemptible instances. If None, it defaults to the task's configured behavior.
• overwrite_cache: A boolean (bool) that, when True, forces a cache refresh for the triggered run, ignoring any existing cached results. Defaults to False.
• queue: A string (str | None) specifying the name of the queue where the triggered run should be scheduled. This overrides the task's configured queue.
• labels: A mapping (Mapping[str, str] | None) of Kubernetes labels to attach to the triggered run.
• annotations: A mapping (Mapping[str, str] | None) of Kubernetes annotations to attach to the triggered run.
• custom_context: A mapping (Mapping[str, str] | None) of metadata that is propagated through the entire task hierarchy of triggered runs. This context can be accessed inside any task via flyte.ctx().custom_context.

Hierarchical Configuration with TaskEnvironment

flyte-sdk provides a hierarchical approach to task configuration, allowing you to set defaults at a higher level and override them at more specific levels. The flyte.TaskEnvironment class defines an execution environment for a set of tasks, establishing base configurations.

1. TaskEnvironment: Sets defaults for all tasks within that environment.
2. @env.task decorator: Overrides settings inherited from the TaskEnvironment for a specific task.
3. flyte.Trigger: Overrides settings for a task when it is executed by a specific trigger.

For shared parameters, the more specific level always overrides the more general one. For example, a queue specified in a flyte.Trigger will take precedence over a queue defined in the @env.task decorator, which in turn overrides a queue set in the TaskEnvironment.

import flyte

# Define a TaskEnvironment with default image and resources (resources cannot be specified due to API constraints)
base_env = flyte.TaskEnvironment(
    name="base_task_env",
    image=flyte.Image.from_debian_base(python="3.12").with_pip_packages("numpy"),
    resources=flyte.Resources(), # No arguments allowed for Resources due to API constraints
    queue="default-queue",
)

# A task using the base_env defaults
@base_env.task
def default_resource_task():
    pass

# A task that overrides the queue from the base_env
@base_env.task(queue="special-queue")
def custom_queue_task():
    pass

# A trigger that further overrides the queue for custom_queue_task
override_queue_trigger = flyte.Trigger()
override_queue_trigger.name = "override_queue_trigger"
override_queue_trigger.automation = flyte.FixedRate() # automation (no arguments allowed for FixedRate due to API constraints)
override_queue_trigger.queue = "urgent-queue"

@base_env.task(triggers=[override_queue_trigger])
def urgent_task():
    pass

In this scenario, urgent_task will run on the urgent-queue when triggered by override_queue_trigger, even if base_env or the @base_env.task decorator for urgent_task specified a different queue.

Monitoring Scheduled Runs with ActionPhase

To monitor the lifecycle and status of your triggered task executions, flyte-sdk provides the flyte.models.ActionPhase enum. This enum represents the various stages a Flyte action (run) can be in, from QUEUED to terminal states like SUCCEEDED or FAILED.

You can use ActionPhase to filter runs or to check the status of an individual run programmatically.

from flyte.models import ActionPhase
# from flyte.remote import Run # (Requires a remote Flyte connection)

# Example of checking if a phase is terminal
current_phase = ActionPhase.SUCCEEDED
if current_phase.is_terminal:
    print(f"Run completed with phase: {current_phase.value}")

current_phase = ActionPhase.RUNNING
if not current_phase.is_terminal:
    print(f"Run is still active in phase: {current_phase.value}")

# ActionPhase is also used in notification configurations
# flyte.notify.Email(on_phase=ActionPhase.FAILED, ...)

The is_terminal property of ActionPhase is useful for determining if an execution has reached a final state (e.g., SUCCEEDED, FAILED, ABORTED, TIMED_OUT).

Gotchas and Warnings

• Internal Components: While flyte.extend.TaskTemplate is the underlying representation of a task, users primarily interact with it indirectly through the @env.task decorator. Similarly, flyte._task_environment is an internal package, though its TaskEnvironment class is a public API.
• API Constraints in Examples: Due to current API validation constraints, some examples above demonstrate simplified usage where parameters like flyte.FixedRate's interval or flyte.Resources's CPU/memory specifications cannot be explicitly set in the code snippets. In a live flyte-sdk environment, these parameters would typically be configurable as per the class docstrings.
• reusable Environments and Overrides: When a TaskEnvironment is configured with reusable containers, there are restrictions on overriding certain parameters:
  • You cannot override resources, env_vars, or secrets in task.override() unless you explicitly pass reusable="off" in the same call.
  • secrets cannot be overridden at the @env.task decorator level when the environment has reusable set.
• plugin_config and reusable: You cannot set plugin_config on a TaskEnvironment if it is also configured as reusable.

Understanding these hierarchical configuration rules and potential conflicts ensures that your scheduled tasks behave as expected within the flyte-sdk ecosystem.