Launch plans, schedules, and fixed inputs

Launch plans in flytekit provide a mechanism to parameterize workflow executions, apply default or fixed inputs, and define schedules or triggers. While every workflow is registered with a default launch plan, you can create custom launch plans to handle specific execution scenarios, such as recurring jobs or executions with pre-configured parameters.

Creating Launch Plans

When you define a workflow, flytekit automatically creates a default launch plan for it. You can retrieve this or create a custom one using the LaunchPlan.get_or_create method.

from flytekit import workflow, LaunchPlan

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

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

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

Internally, LaunchPlan.get_or_create manages a cache (LaunchPlan.CACHE) to ensure that multiple calls for the same workflow or name return the same instance. If you provide a name, flytekit uses LaunchPlan.create to initialize the object, which transforms your Python native inputs into Flyte-compatible literals using translate_inputs_to_literals.

Default vs. Fixed Inputs

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

• Default Inputs: Passed via the default_inputs argument. These values are used if no value is provided at launch time, but they can be overridden by the user.
• Fixed Inputs: Passed via the fixed_inputs argument. These values are immutable for any execution triggered by that specific launch plan.

In the LaunchPlan constructor, any keys present in fixed_inputs are automatically removed from the parameters map (which holds the interface for user-provided inputs) to ensure they cannot be modified.

Scheduling Executions

You can automate workflow executions by attaching a schedule to a launch plan. flytekit supports two primary scheduling mechanisms: CronSchedule and FixedRate.

Cron Schedules

Use CronSchedule when you need to run a workflow based on a cron expression.

from flytekit import LaunchPlan
from flytekit.core.schedule import CronSchedule

hourly_lp = LaunchPlan.get_or_create(
    name="hourly_execution",
    workflow=my_workflow,
    schedule=CronSchedule(
        schedule="0 * * * *",  # Runs at the start of every hour
        kickoff_time_input_arg="kickoff_time" # Optional: maps the schedule time to a workflow input
    ),
    default_inputs={"a": 1, "b": "cron"}
)

The CronSchedule class validates the expression using the croniter library. It supports standard 5-field cron formats and common aliases like @hourly or @daily.

Fixed Rate Schedules

Use FixedRate for executions that should occur at a consistent interval.

from datetime import timedelta
from flytekit import LaunchPlan
from flytekit.core.schedule import FixedRate

interval_lp = LaunchPlan.get_or_create(
    name="interval_execution",
    workflow=my_workflow,
    schedule=FixedRate(duration=timedelta(minutes=10)),
    default_inputs={"a": 5, "b": "fixed-rate"}
)

FixedRate translates a datetime.timedelta into a FixedRateUnit (MINUTE, HOUR, or DAY). Note that flytekit does not support a granularity of less than one minute for fixed-rate schedules.

Launch Plan Triggers

The trigger parameter in LaunchPlan.get_or_create provides a newer syntax for specifying how a launch plan is invoked. The OnSchedule class acts as a wrapper for either a CronSchedule or a FixedRate object, conforming to the LaunchPlanTriggerBase protocol.

from flytekit.core.schedule import OnSchedule, CronSchedule

triggered_lp = LaunchPlan.get_or_create(
    name="triggered_lp",
    workflow=my_workflow,
    trigger=OnSchedule(CronSchedule(schedule="*/5 * * * *"))
)

Advanced Usage

Launch Plans in Workflows

Launch plans can be invoked within other workflows just like tasks or sub-workflows. When a launch plan is called inside a workflow, flytekit uses create_and_link_node to add it to the execution graph.

@workflow
def parent_workflow(x: int) -> str:
    # Invoking a launch plan
    return custom_lp(a=x)

Reference Launch Plans

If you need to trigger a launch plan that is already registered on a Flyte cluster without having the original source code available, use ReferenceLaunchPlan or the @reference_launch_plan decorator. This allows you to define the expected interface locally while pointing to the remote entity.

from flytekit import reference_launch_plan

@reference_launch_plan(
    project="flytesnacks",
    domain="development",
    name="my_remote_lp",
    version="v1"
)
def remote_lp(a: int, b: str) -> str:
    ...

The ReferenceLaunchPlan constructor takes the project, domain, name, and version, along with the expected input and output type mappings, to facilitate compilation without a network call to the Flyte Admin service.