Skip to main content

Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, enforce specific input values, and define execution schedules. While every workflow is automatically registered with a default launch plan, you can create custom launch plans to handle recurring jobs or specialized execution configurations.

Creating Launch Plans​

When you define a workflow, flytekit allows you to create a LaunchPlan that wraps it. The most common way to interact with launch plans is through the LaunchPlan.get_or_create method.

If you need a simple entry point for a workflow without any special configuration, you can retrieve the default launch plan:

from flytekit import workflow, LaunchPlan

@workflow
def my_wf(a: int, b: str) -> str:
    return f"{b}: {a}"

# Retrieves the default launch plan (named after the workflow)
default_lp = LaunchPlan.get_or_create(workflow=my_wf)

Internally, LaunchPlan.get_or_create checks a local CACHE (defined in flytekit/core/launch_plan.py) to ensure that multiple calls for the same workflow return the same object, preventing redundant entity creation during serialization.

Parameterizing Inputs​

Launch plans allow you to pre-define inputs in two ways: default_inputs and fixed_inputs.

Default Inputs​

Use default_inputs to provide values that can still be overridden at execution time. This is useful for providing sensible defaults for scheduled runs while allowing manual overrides.

lp_with_defaults = LaunchPlan.get_or_create(
    name="defaults_lp",
    workflow=my_wf,
    default_inputs={"a": 42}
)

Fixed Inputs​

Use fixed_inputs to lock specific parameters. These values cannot be changed when the launch plan is invoked. If a user attempts to provide a different value for a fixed input at runtime, flytekit will ignore it or fail depending on the execution context.

lp_with_fixed = LaunchPlan.get_or_create(
    name="fixed_lp",
    workflow=my_wf,
    fixed_inputs={"b": "fixed-value"}
)

In the LaunchPlan constructor, flytekit ensures that fixed_inputs are removed from the parameters map (the set of inputs exposed to the user) and stored in self._fixed_inputs as a LiteralMap. This separation ensures that the Flyte engine treats them as immutable constants for that specific launch plan.

Scheduling Executions​

Launch plans are the primary way to schedule workflows in Flyte. You can define schedules using either CronSchedule or FixedRate.

Cron Schedules​

CronSchedule supports standard cron expressions. This is ideal for workflows that need to run at specific times (e.g., "every day at midnight").

from flytekit import CronSchedule

daily_lp = LaunchPlan.get_or_create(
    name="daily_report",
    workflow=my_wf,
    schedule=CronSchedule(schedule="0 0 * * *"),
    default_inputs={"a": 1, "b": "daily"}
)

The CronSchedule class in flytekit/core/schedule.py validates the cron expression using the croniter library. It also supports a kickoff_time_input_arg parameter, which allows you to pass the actual time the schedule triggered the execution into one of your workflow's datetime inputs.

Fixed Rate Schedules​

FixedRate is used for intervals, such as "every 10 minutes". It accepts a datetime.timedelta object.

from datetime import timedelta
from flytekit import FixedRate

frequent_lp = LaunchPlan.get_or_create(
    name="frequent_check",
    workflow=my_wf,
    schedule=FixedRate(duration=timedelta(minutes=10))
)

Note that FixedRate schedules have a minimum granularity of one minute. The _translate_duration method in FixedRate automatically converts your timedelta into the appropriate FixedRateUnit (MINUTE, HOUR, or DAY) required by the Flyte IDL.

Triggers (Alpha)​

Flytekit also supports a trigger argument in LaunchPlan.get_or_create. This is a newer, more flexible way to define how a launch plan is invoked. Currently, the OnSchedule class acts as a wrapper for the existing schedule types:

from flytekit import OnSchedule

triggered_lp = LaunchPlan.get_or_create(
    name="triggered_lp",
    workflow=my_wf,
    trigger=OnSchedule(FixedRate(duration=timedelta(hours=1)))
)

Local Execution and Composition​

Launch plans are callable objects. When you call a launch plan locally, it behaves like the underlying workflow but merges the saved_inputs (defaults and fixed values) with any keyword arguments you provide.

# Locally executes the workflow with a=42 and b="manual"
lp_with_defaults(b="manual")

In flytekit/core/launch_plan.py, the __call__ method detects if it is being called during a compilation phase (e.g., inside another workflow). If so, it uses create_and_link_node to represent the launch plan execution as a node in the workflow graph. This allows you to use launch plans as dependencies in dynamic tasks or as targets for ArrayNode (map tasks).