Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a way to parameterize workflow executions, allowing you to define default or fixed inputs and schedule executions. While every workflow is registered with a default launch plan, creating custom launch plans enables you to reuse the same workflow logic with different configurations or triggers.

Creating Launch Plans

You create a launch plan using the LaunchPlan.get_or_create method. If you don't provide a name, flytekit returns the default launch plan for the workflow. If you provide a name, you can specify additional attributes like inputs and schedules.

from flytekit import workflow, LaunchPlan

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

# Get the default launch plan
default_lp = LaunchPlan.get_or_create(workflow=my_wf)

# Create a named launch plan with custom settings
custom_lp = LaunchPlan.get_or_create(
    name="my_custom_launch_plan",
    workflow=my_wf,
    default_inputs={"a": 10},
    fixed_inputs={"b": "fixed_value"}
)

Internally, LaunchPlan.get_or_create (defined in flytekit/core/launch_plan.py) manages a cache of launch plans to ensure that multiple calls for the same named plan return the same object. If you attempt to create two launch plans with the same name but different configurations, flytekit raises an AssertionError.

Parameterizing Inputs

Launch plans distinguish between default_inputs and fixed_inputs:

Default Inputs: These provide values that are used if no input is provided at execution time. They can be overridden by the user when manually triggering the launch plan.
Fixed Inputs: These values are locked into the launch plan. They cannot be changed at execution time.

When you define a launch plan, flytekit validates these inputs against the workflow's interface. In LaunchPlan.create, the translate_inputs_to_literals function converts your Python native values into Flyte's internal LiteralMap format.

# 'a' can be changed at launch, but 'b' is always "constant"
lp = LaunchPlan.get_or_create(
    name="parameterized_lp",
    workflow=my_wf,
    default_inputs={"a": 42},
    fixed_inputs={"b": "constant"}
)

Scheduling Executions

To automate workflow runs, you can attach a schedule to a launch plan. flytekit supports two primary scheduling mechanisms: CronSchedule and FixedRate.

Cron Schedules

CronSchedule allows you to define executions using standard cron expressions or aliases like @daily.

from flytekit import LaunchPlan, CronSchedule

daily_lp = LaunchPlan.get_or_create(
    name="daily_execution",
    workflow=my_wf,
    schedule=CronSchedule(schedule="@daily"),
    default_inputs={"a": 1, "b": "daily_run"}
)

The CronSchedule class (in flytekit/core/schedule.py) validates the cron string using the croniter library. It also supports a kickoff_time_input_arg parameter, which allows you to pass the scheduled time into a specific workflow input.

Fixed Rate Schedules

FixedRate is used for intervals that occur at a consistent frequency, such as every 10 minutes.

from datetime import timedelta
from flytekit import LaunchPlan, FixedRate

frequent_lp = LaunchPlan.get_or_create(
    name="every_ten_minutes",
    workflow=my_wf,
    schedule=FixedRate(duration=timedelta(minutes=10)),
    default_inputs={"a": 5, "b": "interval_run"}
)

Note that FixedRate schedules have a minimum granularity of one minute. The _translate_duration method in FixedRate ensures the timedelta is converted into a valid FixedRateUnit (MINUTE, HOUR, or DAY).

Using Launch Plans in Dynamic Workflows

When you need to trigger a launch plan from within a @dynamic task, you must explicitly declare the dependency. Because dynamic tasks are compiled at runtime, Flyte needs to know which launch plans must be registered beforehand.

Use the node_dependency_hints parameter in the @dynamic decorator to include your launch plan:

from flytekit import dynamic

@dynamic(node_dependency_hints=[custom_lp])
def dynamic_trigger():
    # Trigger the launch plan multiple times
    return [custom_lp(a=i) for i in range(5)]

When a launch plan is called inside a compilation context (like a dynamic task or another workflow), the LaunchPlan.__call__ method invokes create_and_link_node, which integrates the launch plan execution into the workflow graph.

Reference Launch Plans

If you need to trigger a launch plan that is already registered in a different Flyte project or domain, use ReferenceLaunchPlan or the @reference_launch_plan decorator. This allows you to reference the entity without redefining its logic, provided you supply the expected interface.

from flytekit import reference_launch_plan

@reference_launch_plan(
    project="other_project",
    domain="production",
    name="existing_lp",
    version="v1"
)
def remote_lp(a: int, b: str) -> str:
    ...

The ReferenceLaunchPlan class (in flytekit/core/launch_plan.py) acts as a pointer and does not initiate network calls during local definition, ensuring that cross-project dependencies are handled during the registration phase on the Flyte backend.

Creating Launch Plans​

Parameterizing Inputs​

Scheduling Executions​

Cron Schedules​

Fixed Rate Schedules​

Using Launch Plans in Dynamic Workflows​

Reference Launch Plans​