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, set up schedules, and configure notifications. While every workflow is registered with a default launch plan, creating custom launch plans enables you to reuse the same workflow logic across different execution contexts.

Creating Launch Plans

You create launch plans using the LaunchPlan.get_or_create method. If you do not provide a name, flytekit returns the default launch plan for the workflow. If you specify additional properties like schedules or fixed inputs, you must provide a unique name.

Default Launch Plans

A default launch plan uses the workflow's signature to determine its inputs and has no additional configurations. This is commonly used when you need to reference a workflow within a dynamic task.

from flytekit import workflow, dynamic, 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(my_workflow)

@dynamic(node_dependency_hints=[default_lp])
def launch_dynamically():
    # The launch plan can be called like a function within dynamic tasks
    return [default_lp(a=i, b="val") for i in range(10)]

Internally, LaunchPlan.get_or_create caches launch plans by name to prevent duplicate entities. When no name is provided, it uses the workflow's name as the key in LaunchPlan.CACHE.

Fixed and Default Inputs

Launch plans allow you to specialize a workflow by pre-defining its inputs.

• Default Inputs: These provide values that can be overridden at execution time.
• Fixed Inputs: These values are locked and cannot be changed when the launch plan is triggered.

from flytekit import workflow, LaunchPlan

@workflow
def process_data(region: str, threshold: float) -> float:
    ...

# Create a launch plan with a fixed region and a default threshold
region_lp = LaunchPlan.get_or_create(
    workflow=process_data,
    name="us_east_lp",
    default_inputs={"threshold": 0.5},
    fixed_inputs={"region": "us-east-1"}
)

When LaunchPlan.create is called, it translates these Python values into Flyte literals. Fixed inputs are stored in self._fixed_inputs (a LiteralMap), and flytekit ensures they are removed from the parameter_map so they cannot be overridden during execution.

Scheduling Executions

You can automate workflow runs by attaching a schedule to a launch plan. flytekit supports two primary types of schedules: CronSchedule and FixedRate.

Cron Schedules

CronSchedule uses a cron expression to define execution intervals. It supports standard 5-field cron formats and aliases like @daily or @hourly.

from flytekit import workflow, LaunchPlan, CronSchedule
from datetime import datetime

@workflow
def daily_job(kickoff_time: datetime):
    ...

daily_lp = LaunchPlan.get_or_create(
    workflow=daily_job,
    name="daily_cron_lp",
    schedule=CronSchedule(
        schedule="0 0 * * *",  # Runs every day at midnight
        kickoff_time_input_arg="kickoff_time"
    )
)

The kickoff_time_input_arg parameter allows the workflow to receive the exact time the schedule triggered the execution.

Fixed Rate Schedules

FixedRate schedules trigger executions at a consistent frequency defined by a datetime.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))
)

Note that FixedRate schedules in flytekit do not support granularity of less than one minute. The FixedRate._translate_duration method validates the duration and converts it into DAY, HOUR, or MINUTE units for the Flyte backend.

Reference Launch Plans

If you need to trigger a launch plan that is already registered in a different project or domain, use ReferenceLaunchPlan. This allows you to define the interface locally without needing the original workflow source code.

from flytekit import ReferenceLaunchPlan

existing_lp = ReferenceLaunchPlan(
    project="shared_project",
    domain="production",
    name="data_ingestion_lp",
    version="v1",
    inputs={"data_path": str},
    outputs={"result_count": int}
)

Alternatively, you can use the @reference_launch_plan decorator to define the interface using a function signature:

from flytekit import reference_launch_plan

@reference_launch_plan(
    project="shared_project",
    domain="production",
    name="data_ingestion_lp",
    version="v1"
)
def existing_lp(data_path: str) -> int:
    ...

Execution Behavior

When you call a LaunchPlan object, its behavior depends on the context:

1. Compilation Context: If called inside a workflow or dynamic task, flytekit uses create_and_link_node to add the launch plan as a node in the workflow graph.
2. Local Execution: If called directly in a Python script, it forwards the call to the underlying self.workflow, merging the saved_inputs (defaults and fixed values) with any keyword arguments provided at the call site.

Launch plans only support keyword arguments during execution to ensure clarity when overriding default inputs.