Launch plans, schedules, and fixed inputs

Launch Plans

Launch plans are the gateway to workflow execution in Flyte. While a workflow defines what your computation does, a launch plan controls how it runs — specifying default values, locking inputs that shouldn't change, connecting schedules, and attaching metadata like labels and notifications.

Every workflow registered with Flyte automatically gets a default launch plan. You create custom launch plans to add behavior beyond the basics.

Creating a Launch Plan

The primary entry point is LaunchPlan.get_or_create(). Call it with your workflow:

from flytekit import workflow, LaunchPlan

@workflow
def process_data(filename: str, threshold: int = 10) -> str:
    ...

# Get the default launch plan (no extra configuration)
lp = LaunchPlan.get_or_create(workflow=process_data)

The default launch plan takes its name from the workflow (process_data in this case) and uses any default values from the function signature. It cannot have schedules, fixed inputs, or other custom settings — attempting to add them raises a ValueError.

To add configuration, supply a unique name:

from flytekit import workflow, LaunchPlan

@workflow
def process_data(filename: str, threshold: int = 10) -> str:
    ...

lp = LaunchPlan.get_or_create(
    workflow=process_data,
    name="process-data-production",
    default_inputs={"filename": "daily.csv"},
)

The name must be unique across your Flyte deployment — project, domain, version, and this name form the primary key. If you create two launch plans with the same name for different workflows, get_or_create raises an AssertionError.

Default vs. Fixed Inputs

Launch plans separate inputs into two categories with different behaviors:

Default inputs provide values when the caller doesn't supply them, but callers can override them:

from flytekit import workflow, LaunchPlan

@workflow
def train_model(features_path: str, epochs: int = 5, lr: float = 0.001) -> float:
    ...

lp = LaunchPlan.get_or_create(
    workflow=train_model,
    name="train-defaults",
    default_inputs={"epochs": 20, "lr": 0.01},
)

# Caller can override defaults
result = lp(epochs=10)  # epochs=10, lr=0.01

Default values come from two sources — the workflow function signature and the default_inputs dict — with default_inputs taking precedence. In the example above, lr defaults to 0.01 from default_inputs, not 0.001 from the signature.

Fixed inputs cannot be changed at launch time. They are stripped from the launch plan's parameter map during construction. If a caller passes a fixed input, Flyte ignores it:

from flytekit import workflow, LaunchPlan

@workflow
def train_model(features_path: str, model_type: str) -> float:
    ...

lp = LaunchPlan.get_or_create(
    workflow=train_model,
    name="train-xgboost",
    fixed_inputs={"model_type": "xgboost"},
)

# model_type is locked to "xgboost" — callers cannot change it
result = lp(features_path="s3://data/train.csv")
# Passing model_type here would be silently ignored

Internally, fixed inputs are stored as a LiteralMap (protobuf-serialized) and removed from the parameters property. The __call__ method in LaunchPlan merges saved inputs with caller kwargs before passing them to the workflow.

Schedules

Attach a schedule to run your launch plan on a cron-like interval. Use CronSchedule or FixedRate:

from flytekit import workflow, LaunchPlan
from flytekit.schedule import CronSchedule
from datetime import timedelta
from flytekit.schedule import FixedRate

CronSchedule accepts either a cron alias (hourly, daily, weekly, monthly, yearly) or a croniter-parseable expression:

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

@workflow
def daily_report(date: datetime):
    ...

lp = LaunchPlan.get_or_create(
    workflow=daily_report,
    name="daily-report",
    schedule=CronSchedule(
        schedule="daily",
        kickoff_time_input_arg="date",
    ),
)

The schedule parameter accepts these aliases: hourly, hours, @hourly, daily, days, @daily, weekly, weeks, @weekly, monthly, months, @monthly, annually, @annually, yearly, years, @yearly. Or use a cron expression like "0 */6 * * *" for every 6 hours.

The kickoff_time_input_arg injects the scheduled run time into your workflow. Your workflow must have a matching input parameter.

You can also apply an offset using ISO 8601 duration format (offset="P1D" for one day) to shift when runs trigger.

Note: The cron_expression parameter is deprecated. Use schedule instead.

FixedRate schedules by elapsed time between runs:

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

@workflow
def monitor_pipeline(run_id: str):
    ...

lp = LaunchPlan.get_or_create(
    workflow=monitor_pipeline,
    name="monitor-every-10-min",
    schedule=FixedRate(duration=timedelta(minutes=10)),
)

The minimum granularity is one minute — timedelta(seconds=30) raises an AssertionError.

Notifications

Notify external systems when a workflow reaches a terminal phase. Import notification types from flytekit.models.notification:

from flytekit import workflow, LaunchPlan
from flytekit.models.notification import Email
from flytekit.models.core.execution import WorkflowExecutionPhase

@workflow
def etl_pipeline():
    ...

lp = LaunchPlan.get_or_create(
    workflow=etl_pipeline,
    name="etl-with-alerts",
    notifications=[
        Email(
            phases=[WorkflowExecutionPhase.SUCCEEDED, WorkflowExecutionPhase.FAILED],
            recipients_email=["data-team@company.com"],
        ),
    ],
)

Valid phases include SUCCEEDED, FAILED, ABORTED, and TIMED_OUT. You can also use PagerDuty and Slack from the same module.

Execution Control Options

Several options control how Flyte executes your launch plan:

max_parallelism limits concurrent task nodes across the workflow. This is useful for rate-limiting resources or ensuring fairness when many workflows run simultaneously. Map tasks count as a single unit:

lp = LaunchPlan.get_or_create(
    workflow=process_data,
    name="rate-limited",
    max_parallelism=5,
)

overwrite_cache forces cache invalidation. By default (False), cached outputs from previous runs are reused when inputs match. Set to True to always recompute:

lp = LaunchPlan.get_or_create(
    workflow=process_data,
    name="fresh-run",
    overwrite_cache=True,
)

auto_activate registers the launch plan in an activated state. Default is False, meaning you must manually activate it before scheduled runs trigger:

lp = LaunchPlan.get_or_create(
    workflow=process_data,
    name="auto-activate-me",
    auto_activate=True,
)

Labels and Annotations

Attach metadata to executions created by this launch plan. Labels are key-value string pairs; annotations are arbitrary key-value metadata:

from flytekit import workflow, LaunchPlan
from flytekit.models.common import Labels, Annotations

lp = LaunchPlan.get_or_create(
    workflow=process_data,
    name="labeled-run",
    labels=Labels({"team": "ml-platform", "env": "production"}),
    annotations=Annotations({"cost-center": "ml-ops", "owner": "alice"}),
)

Offloaded Data Location

For workflows that produce large outputs, specify a remote storage location for offloaded data using raw_output_data_config:

from flytekit import workflow, LaunchPlan
from flytekit.models.common import RawOutputDataConfig

lp = LaunchPlan.get_or_create(
    workflow=process_data,
    name="offloaded-results",
    raw_output_data_config=RawOutputDataConfig(output_location_prefix="s3://my-bucket/results/"),
)

Supported prefixes include s3://, gs://, and file://.

Security Context

Specify the IAM role and Kubernetes service account for execution:

from flytekit import workflow, LaunchPlan
from flytekit.models.security import SecurityContext, Identity

lp = LaunchPlan.get_create(
    workflow=process_data,
    name="secure-run",
    security_context=SecurityContext(
        run_as=Identity(
            iam_role="arn:aws:iam::123456789:role/flyte-execution-role",
            k8s_service_account="flyte-executors",
        )
    ),
)

Note: AuthRole is deprecated. Use security_context instead.

Using Launch Plans in Dynamic Tasks

When you invoke a launch plan inside a @dynamic task, Flyte requires the launch plan to be registered on FlyteAdmin beforehand. Use node_dependency_hints to ensure registration:

from flytekit import workflow, LaunchPlan, dynamic

@workflow
def sub_workflow(item: str) -> str:
    ...

sub_lp = LaunchPlan.get_or_create(sub_workflow)

@dynamic(node_dependency_hints=[sub_lp])
def run_all(items: List[str]):
    return [sub_lp(item=item) for item in items]

Without node_dependency_hints, the dynamic task would fail at runtime because the sub-launch plan wouldn't exist on FlyteAdmin yet.

Reference Launch Plans

Point to a launch plan that already exists on your Flyte deployment without registering a new one. This is useful for referencing shared infrastructure:

from flytekit import reference_launch_plan

@reference_launch_plan(project="my-org", domain="production", name="shared-transform", version="1.0.0")
def shared_transform(features: str) -> str:
    ...

The decorated function provides the interface — no implementation is needed. ReferenceLaunchPlan does not make network calls at creation time; mismatches are caught during registration.

Cloning Launch Plans

Create a modified copy of an existing launch plan with clone_with():

lp_v2 = lp.clone_with(
    name="process-data-v2",
    default_inputs={"threshold": 20},
)

Any parameter not passed to clone_with() inherits from the original. This is useful for creating variants with different defaults or schedules without rebuilding from scratch.

Calling Launch Plans

You call a launch plan like a function, passing inputs as keyword arguments:

result = lp(features_path="s3://data/train.csv", threshold=15)

During compilation (e.g., inside another workflow), calling a launch plan creates a node. During execution, it invokes the underlying workflow directly, merging in saved defaults and fixed inputs.

Registration

Launch plans are collected in FlyteEntities.entities during __init__. When you register your workflow code with pyflyte register, all launch plans in this list are registered alongside it.