Skip to main content

Launch plans, schedules, and fixed inputs

A Flyte workflow defines the structural directed acyclic graph (DAG) of computation tasks and interfaces, but running that workflow often requires varying schedules, locked parameters, or specific resource allocations. In flytekit, a LaunchPlan serves as the execution blueprint for a workflow, decoupling parameter configuration, execution cadences, notifications, and security settings from the core workflow definition.

Defining Launch Plans​

When you define a @workflow, flytekit implicitly creates a default launch plan matching the workflow's name, its native default inputs, and its default execution options. However, when you need multiple entry points with customized behaviors—such as automated production runs or locked input configurations—you define named launch plans.

Creating Named and Default Launch Plans​

Use LaunchPlan.get_or_create(...) to construct and register launch plans. The factory method ensures uniqueness and avoids duplicate instantiations within the same session.

from flytekit import LaunchPlan, task, workflow


@task
def calculate_metrics(threshold: float, dataset_name: str) -> float:
    return threshold * len(dataset_name)


@workflow
def metrics_wf(threshold: float = 0.5, dataset_name: str = "default_set") -> float:
    return calculate_metrics(threshold=threshold, dataset_name=dataset_name)


# 1. Access the default launch plan (unnamed, uses workflow defaults)
default_lp = LaunchPlan.get_or_create(workflow=metrics_wf)

# 2. Create a named launch plan with specialized parameters
prod_lp = LaunchPlan.get_or_create(
    workflow=metrics_wf,
    name="metrics_wf_production",
    default_inputs={"threshold": 0.8},
    fixed_inputs={"dataset_name": "production_dataset"},
)

Unnamed vs. Named Launch Plan Rules​

If you invoke LaunchPlan.get_or_create(workflow=...) without passing name, flytekit assumes you are referencing the default launch plan. Supplying any parameterization (such as default_inputs, fixed_inputs, schedule, notifications, or security_context) without specifying name raises a ValueError:

ValueError: Only named launchplans can be created that have other properties. Drop the name if you want to create a default launchplan. Default launchplans cannot have any other associations

In-Memory Caching and Identity Validation​

LaunchPlan.CACHE indexes all instantiated launch plans by name in memory. When LaunchPlan.get_or_create is called with an existing name:

  1. It verifies that the associated workflow matches cached_outputs["_workflow"]. If names match across different workflows, flytekit raises an AssertionError.
  2. It compares all configured attributes (schedule, notifications, default_inputs, labels, annotations, raw_output_data_config, max_parallelism, security_context, overwrite_cache, auto_activate) against the cached plan. If any attribute differs, flytekit raises an AssertionError preventing accidental silent overrides.

Default vs. Fixed Inputs​

Launch plans support two mechanisms for pre-populating workflow inputs: default_inputs and fixed_inputs.

Workflow Signature:       (a: int, b: str, c: float)
                                │        │         │
LaunchPlan.get_or_create(      │        │         │
    default_inputs={"b": "foo"}─┴────────┘         │
    fixed_inputs={"c": 1.23}───────────────────────┘
                                │
                        Execution Inputs:
                        - "a" (Required at launch)
                        - "b" (Optional, defaults to "foo", overridable)
                        - "c" (Locked to 1.23, NOT in ParameterMap)

Parameterizing Inputs​

default_inputs provide fallback values that can still be overridden when triggering an execution. fixed_inputs are immutable values locked at launch plan creation time.

from flytekit import LaunchPlan, task, workflow


@task
def process_data(region: str, sample_size: int, debug: bool) -> str:
    return f"{region}_{sample_size}_{debug}"


@workflow
def data_pipeline(region: str, sample_size: int = 100, debug: bool = False) -> str:
    return process_data(region=region, sample_size=sample_size, debug=debug)


# Lock "region" so it cannot be altered; default "sample_size" to 5000 (can be overridden)
custom_lp = LaunchPlan.get_or_create(
    workflow=data_pipeline,
    name="us_east_pipeline",
    default_inputs={"sample_size": 5000},
    fixed_inputs={"region": "us-east-1"},
)

Internal Implementation: ParameterMap vs. LiteralMap​

Under the hood in flytekit/core/launch_plan.py:

  • default_inputs merge into wf_signature_parameters using transform_inputs_to_parameters(ctx, temp_interface). They take higher precedence than workflow signature defaults and remain exposed via LaunchPlan.parameters (a ParameterMap).
  • fixed_inputs are translated directly into Flyte IDL literals using translate_inputs_to_literals(...) and stored in LaunchPlan.fixed_inputs (a LiteralMap).
  • In LaunchPlan.__init__, any input key found in fixed_inputs.literals is explicitly filtered out of the parameter mapping:
from flytekit.models.interface import ParameterMap

# In LaunchPlan.__init__, fixed inputs are excluded from the ParameterMap
filtered_parameters = {k: v for k, v in parameters.parameters.items() if k not in fixed_inputs.literals}
launch_plan_parameters = ParameterMap(parameters=filtered_parameters)

Because fixed inputs are stripped from the launch plan's parameter map, external callers and the Flyte UI/API will not accept user overrides for those arguments at execution time.

Scheduling and Automation​

flytekit provides two primary schedule types in flytekit.core.schedule: CronSchedule and FixedRate. Schedules are attached to launch plans via the schedule parameter and can optionally activate immediately on registration using auto_activate=True.

Cron-Based Schedules (CronSchedule)​

CronSchedule supports standard 5-field cron syntax or aliases (e.g., @hourly, @daily, @weekly, @monthly, @yearly, hourly, daily). Expressions are validated locally using croniter.

from datetime import datetime
from flytekit import LaunchPlan, task, workflow
from flytekit.core.schedule import CronSchedule


@task
def process_batch(kickoff_time: datetime, env: str) -> str:
    return f"Executed for {kickoff_time.isoformat()} in {env}"


@workflow
def scheduled_batch_wf(kickoff_time: datetime, env: str = "production") -> str:
    return process_batch(kickoff_time=kickoff_time, env=env)


# Runs every day at midnight UTC, passing the execution trigger time to `kickoff_time`
daily_cron_lp = LaunchPlan.get_or_create(
    workflow=scheduled_batch_wf,
    name="daily_batch_midnight",
    schedule=CronSchedule(
        schedule="0 0 * * *",
        offset="PT0S",  # ISO 8601 duration offset
        kickoff_time_input_arg="kickoff_time",
    ),
    auto_activate=True,
)

Note: Setting the deprecated parameter cron_expression on CronSchedule raises an AssertionError. Always pass cron expressions or aliases to schedule.

Fixed Interval Schedules (FixedRate)​

FixedRate triggers executions at fixed time intervals using a datetime.timedelta. The duration is automatically mapped internally to standard units (DAY, HOUR, or MINUTE).

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


@task
def heartbeat(kickoff_time: datetime) -> str:
    return f"Heartbeat at {kickoff_time}"


@workflow
def heartbeat_wf(kickoff_time: datetime) -> str:
    return heartbeat(kickoff_time=kickoff_time)


# Runs every 15 minutes
heartbeat_lp = LaunchPlan.get_or_create(
    workflow=heartbeat_wf,
    name="heartbeat_15min",
    schedule=FixedRate(
        duration=timedelta(minutes=15),
        kickoff_time_input_arg="kickoff_time",
    ),
    auto_activate=True,
)

FixedRate requires a minimum granularity of one minute. If duration.microseconds != 0 or duration.seconds % 60 != 0, FixedRate._translate_duration raises an AssertionError.

Dynamic Kickoff Time Injection​

When you supply kickoff_time_input_arg="kickoff_time" in CronSchedule or FixedRate, the Flyte scheduler injects the exact timestamp of the scheduled trigger event directly into the designated workflow input argument.

Execution Options, Notifications, and Security​

Launch plans allow you to customize infrastructure configurations, alerts, and access controls for downstream runs.

from flytekit import Email, LaunchPlan, PagerDuty, Slack, WorkflowExecutionPhase, task, workflow
from flytekit.models.common import Annotations, Labels, RawOutputDataConfig
from flytekit.models.security import Identity, SecurityContext


@task
def critical_task(v: int) -> int:
    return v * 2


@workflow
def mission_critical_wf(v: int) -> int:
    return critical_task(v=v)


managed_lp = LaunchPlan.get_or_create(
    workflow=mission_critical_wf,
    name="mission_critical_lp",
    # Notification alerts on terminal phases
    notifications=[
        Email(
            phases=[WorkflowExecutionPhase.FAILED, WorkflowExecutionPhase.TIMED_OUT],
            recipients_email=["ops-team@example.com"],
        ),
        Slack(
            phases=[WorkflowExecutionPhase.SUCCEEDED],
            recipients_email=["slack-webhook-integration@example.com"],
        ),
        PagerDuty(
            phases=[WorkflowExecutionPhase.FAILED],
            recipients_email=["pagerduty-alert@example.com"],
        ),
    ],
    # Concurrency and cache behavior
    max_parallelism=10,
    overwrite_cache=True,
    # Execution metadata
    labels=Labels({"environment": "production", "team": "data"}),
    annotations=Annotations({"description": "High-priority scheduled run"}),
    # Offloaded raw data storage destination
    raw_output_data_config=RawOutputDataConfig(output_location_prefix="s3://my-company-bucket/raw_data/"),
    # Identity and IAM / K8s service account execution context
    security_context=SecurityContext(
        run_as=Identity(
            iam_role="arn:aws:iam::123456789012:role/DataWorkflowExecutionRole",
            k8s_service_account="workflow-runner-sa",
        )
    ),
)

Notification Constraints​

Classes in flytekit.core.notification (Email, Slack, PagerDuty) validate that provided phases belong strictly to terminal states:

  • WorkflowExecutionPhase.ABORTED
  • WorkflowExecutionPhase.FAILED
  • WorkflowExecutionPhase.SUCCEEDED
  • WorkflowExecutionPhase.TIMED_OUT

Passing an empty phase list or non-terminal phases raises an AssertionError.

Executing and Composing Launch Plans​

Launch plans can be invoked directly inside local Python environments, nested inside other workflows, or triggered dynamically inside @dynamic workflows.

Local Execution vs. Compilation Call Semantics​

When calling a launch plan via lp(...), flytekit branches depending on whether an active compilation_state exists in FlyteContext:

# Internal branching logic in LaunchPlan.__call__
inputs = self.saved_inputs
inputs.update(kwargs)
if ctx.compilation_state is not None:
    return create_and_link_node(ctx, entity=self, **inputs)
else:
    return self.workflow(*args, **inputs)

Rule: Launch plan executions strictly require keyword arguments. Passing positional arguments (e.g., prod_lp(10)) raises an AssertionError.

# Calling launch plan in local execution
result = prod_lp(threshold=0.95)  # dataset_name is pre-filled by fixed_inputs

Nesting Launch Plans in Workflows​

You can call a launch plan as a sub-node inside a parent workflow:

@workflow
def parent_workflow(threshold: float) -> float:
    # Invokes the launch plan node with fixed dataset_name and custom threshold
    return prod_lp(threshold=threshold)

Using Launch Plans in Dynamic Tasks​

When running launch plans inside a @dynamic task, FlyteAdmin must already have the referenced launch plans registered. Declare them in node_dependency_hints so serialization includes their references:

from flytekit import dynamic


@dynamic(node_dependency_hints=[prod_lp])
def dynamic_orchestrator(thresholds: list[float]) -> list[float]:
    results = []
    for t in thresholds:
        results.append(prod_lp(threshold=t))
    return results

Referencing Pre-Registered Remote Launch Plans​

To invoke a launch plan deployed in another project, domain, or version without re-declaring its Python workflow implementation, use ReferenceLaunchPlan or the @reference_launch_plan decorator from flytekit.core.launch_plan.

from flytekit import reference_launch_plan, workflow


@reference_launch_plan(
    project="shared_services",
    domain="production",
    name="customer_etl_lp",
    version="v1.2.0",
)
def remote_customer_etl(customer_id: str, limit: int) -> int: ...


@workflow
def aggregator_workflow(cust_id: str) -> int:
    return remote_customer_etl(customer_id=cust_id, limit=100)

ReferenceLaunchPlan creates a pointer via LaunchPlanReference(project, domain, name, version). It does not perform network round-trips to FlyteAdmin during declaration time; interface validation occurs during registration and execution binding.