Skip to main content

Launch plans, schedules, and fixed inputs

Launch plans in flytekit allow you to define how a workflow should be executed, including pre-defined inputs, schedules, and notifications. While every workflow is registered with a default launch plan, creating custom launch plans enables you to parameterize executions for different environments or recurring automation.

Creating Launch Plans​

When you want to execute a workflow with a specific set of configurations without modifying the workflow definition itself, you use the LaunchPlan class. The recommended way to instantiate a launch plan is through LaunchPlan.get_or_create.

Default Launch Plans​

If you need a basic launch plan that uses the workflow's default parameters and has no special triggers, you can retrieve it by passing only the workflow object.

from flytekit import workflow, LaunchPlan

@workflow
def my_wf(a: int, b: str = "default"):
    ...

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

Internally, LaunchPlan.get_or_create checks a local CACHE in the LaunchPlan class. If a name is not provided, it defaults to the workflow's name and calls get_default_launch_plan, which extracts parameters from the workflow's python_interface.

Parameterizing with Default and Fixed Inputs​

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

  • default_inputs: These values are used if no input is provided during execution, but they can be overridden by the user.
  • fixed_inputs: These values are "baked into" the launch plan and cannot be changed at execution time.
from flytekit import workflow, LaunchPlan

@workflow
def data_pipeline(region: str, threshold: float):
    ...

# A launch plan for a specific region with a default threshold
region_lp = LaunchPlan.get_or_create(
    workflow=data_pipeline,
    name="us_east_pipeline",
    default_inputs={"threshold": 0.5},
    fixed_inputs={"region": "us-east-1"}
)

When LaunchPlan.create is called (via get_or_create), it uses transform_inputs_to_parameters to merge the workflow's signature with your default_inputs. For fixed_inputs, it uses translate_inputs_to_literals to convert Python values into Flyte's internal LiteralMap. The LaunchPlan constructor then ensures that any key present in fixed_inputs is removed from the parameters map, effectively making them non-overridable.

Scheduling Executions​

To run a workflow automatically at specific intervals, you attach a trigger to the launch plan using the OnSchedule class. flytekit supports both cron-based and fixed-rate schedules.

Cron Schedules​

CronSchedule supports standard cron expressions or aliases like @daily.

from flytekit import workflow, LaunchPlan
from flytekit.core.schedule import CronSchedule, OnSchedule

@workflow
def daily_job():
    ...

daily_lp = LaunchPlan.get_or_create(
    workflow=daily_job,
    name="daily_midnight_job",
    trigger=OnSchedule(
        schedule=CronSchedule(schedule="0 0 * * *")
    )
)

The CronSchedule class validates your expression using the croniter library. It also supports a kickoff_time_input_arg parameter. If your workflow accepts a datetime input, you can pass the name of that argument to kickoff_time_input_arg, and Flyte will automatically inject the time the schedule was triggered into that workflow input.

Fixed Rate Schedules​

FixedRate is used for simple intervals, such as "every 10 minutes".

from datetime import timedelta
from flytekit import workflow, LaunchPlan
from flytekit.core.schedule import FixedRate, OnSchedule

@workflow
def heartbeat():
    ...

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

The FixedRate class includes a _translate_duration method that converts a Python timedelta into Flyte's supported units: MINUTE, HOUR, or DAY. Note that flytekit enforces a minimum granularity of one minute; attempting to use seconds or microseconds will result in an AssertionError.

Execution and Local Behavior​

Launch plans are callable objects. When you call a launch plan in a local Python script, it behaves like the underlying workflow but incorporates the saved_inputs (the combination of default and fixed inputs).

# Local execution
# 'region' is fixed to 'us-east-1', 'threshold' defaults to 0.5
region_lp(threshold=0.7)

In the __call__ implementation, flytekit checks the FlyteContext. If it's in a compilation state (e.g., when being used inside another workflow or during registration), it creates a node in the workflow graph using create_and_link_node. Otherwise, it simply forwards the call to self.workflow(*args, **inputs).

Reference Launch Plans​

If you need to trigger a launch plan that is already registered on a Flyte cluster (perhaps managed by a different team or project), use ReferenceLaunchPlan.

from flytekit import ReferenceLaunchPlan

remote_lp = ReferenceLaunchPlan(
    project="shared_project",
    domain="production",
    name="standard_cleanup_lp",
    version="v1",
    inputs={"cutoff": datetime.datetime},
    outputs={}
)

This class inherits from ReferenceEntity. It doesn't contain the workflow logic itself but provides the interface required for flytekit to compile workflows that depend on it. At registration time, Flyte Admin verifies that the interface you defined matches the actual interface of the remote launch plan.