Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, apply fixed or default inputs, and define recurring schedules. While every workflow is registered with a default launch plan, creating custom launch plans allows you to lock down specific configurations for production runs or automated triggers.

Creating Launch Plans

You create launch plans using the LaunchPlan.get_or_create method. This method ensures that launch plans are cached and reused within the same session, preventing duplicate definitions for the same workflow.

Default Launch Plans

If you need a launch plan that simply mirrors the workflow's signature without any modifications, you can retrieve the default one:

from flytekit import workflow, LaunchPlan

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

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

Internally, LaunchPlan.get_default_launch_plan handles this by transforming the workflow's python_interface into a ParameterMap and initializing a LaunchPlan with empty fixed_inputs.

Parameterizing with Default and Fixed Inputs

Launch plans allow you to differentiate between inputs that can be overridden at execution time and those that are immutable.

Default Inputs: Provide a value that is used if the user does not specify one at launch.
Fixed Inputs: Provide a value that cannot be changed at launch time.

from flytekit import workflow, LaunchPlan

@workflow
def training_wf(learning_rate: float, epochs: int, dataset: str) -> float:
    ...

# Create a named launch plan with specific parameters
prod_lp = LaunchPlan.get_or_create(
    workflow=training_wf,
    name="prod_training_lp",
    default_inputs={"epochs": 10},
    fixed_inputs={"dataset": "s3://my-bucket/training-data"}
)

When you call LaunchPlan.get_or_create with a name, flytekit invokes LaunchPlan.create. This method validates that fixed_inputs are translated into LiteralMap objects and removes them from the ParameterMap so they are no longer exposed as tunable parameters in the Flyte UI or CLI.

Scheduling Executions

Flytekit supports recurring executions through the schedule parameter in LaunchPlan.get_or_create. Schedules are implemented using CronSchedule or FixedRate.

Cron Schedules

Use CronSchedule for complex timing requirements. Flytekit supports standard cron expressions and aliases.

from flytekit import workflow, LaunchPlan, CronSchedule

@workflow
def daily_cleanup(target_date: str):
    ...

daily_lp = LaunchPlan.get_or_create(
    workflow=daily_cleanup,
    name="daily_cleanup_lp",
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
    ),
    default_inputs={"target_date": "today"}
)

The CronSchedule class validates the expression using croniter. It also supports a kickoff_time_input_arg, which allows the workflow to receive the exact time the schedule triggered the execution as a datetime input.

Fixed Rate Intervals

For simple periodic tasks, use FixedRate with a timedelta.

from datetime import timedelta
from flytekit import workflow, LaunchPlan, FixedRate

@workflow
def heartbeat():
    ...

heartbeat_lp = LaunchPlan.get_or_create(
    workflow=heartbeat,
    name="heartbeat_lp",
    schedule=FixedRate(duration=timedelta(minutes=10))
)

The FixedRate class automatically translates the timedelta into the appropriate unit (MINUTE, HOUR, or DAY) required by the Flyte IDL. Note that flytekit enforces a minimum granularity of one minute; durations with microseconds or non-zero seconds that don't align to minutes will raise an AssertionError.

Using Launch Plans in Dynamic Workflows

Launch plans can be invoked within other workflows. When used inside a @dynamic task, you must provide the launch plan in the node_dependency_hints to ensure it is correctly registered and available at runtime.

from flytekit import workflow, dynamic, LaunchPlan

@workflow
def sub_wf(x: int):
    ...

sub_lp = LaunchPlan.get_or_create(sub_wf)

@dynamic(node_dependency_hints=[sub_lp])
def dynamic_launcher(n: int):
    # Returns a list of launch plan executions
    return [sub_lp(x=i) for i in range(n)]

When a LaunchPlan is called (__call__), flytekit checks the FlyteContext. During compilation (like inside a dynamic task), it uses create_and_link_node to represent the execution as a node in the workflow graph. During local execution, it simply forwards the call to the underlying workflow using the saved_inputs (defaults and fixed values) merged with any provided keyword arguments.

Referencing Existing Launch Plans

If you need to trigger a launch plan that is already registered in a different project or domain, use ReferenceLaunchPlan or the @reference_launch_plan decorator.

from flytekit import reference_launch_plan

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="core.control_flow.merge_sort.merge_sort_lp",
    version="v1"
)
def remote_lp(sorted_list: list[int]) -> list[int]:
    ...

The ReferenceLaunchPlan class acts as a pointer. It does not perform network calls during initialization but requires you to define the expected interface so flytekit can perform type checking during compilation.

Creating Launch Plans​

Default Launch Plans​

Parameterizing with Default and Fixed Inputs​

Scheduling Executions​

Cron Schedules​

Fixed Rate Intervals​

Using Launch Plans in Dynamic Workflows​

Referencing Existing Launch Plans​