Skip to main content

Launch plans, schedules, and fixed inputs

When deploying workflows to Flyte, registering the bare workflow logic is often not enough. Workflows frequently require different configurations depending on the environment—such as running with locked production parameters on a weekly cron schedule, running nightly with debug notifications, or running with bounded concurrency.

A LaunchPlan in flytekit encapsulates a workflow alongside pre-configured inputs, execution schedules, notification rules, security credentials, and runtime constraints without modifying the underlying workflow definition.

Defining and Creating Launch Plans​

A workflow definition specifies logic and interface types. A LaunchPlan provides an execution specification for that workflow.

Default Launch Plans​

Every workflow automatically has a default launch plan. A default launch plan has no custom name, carries no schedule or notifications, and inherits any default parameter values defined in the workflow function signature.

To retrieve or register the default launch plan for a workflow, call LaunchPlan.get_or_create() with only the workflow argument:

from flytekit import task, workflow
from flytekit.core.launch_plan import LaunchPlan

@task
def process_data(dataset: str, retries: int) -> str:
    return f"Processed {dataset} with {retries} retries"

@workflow
def data_pipeline(dataset: str, retries: int = 3) -> str:
    return process_data(dataset=dataset, retries=retries)

# Retrieve or create the default launch plan
default_lp = LaunchPlan.get_or_create(workflow=data_pipeline)

If you try to pass custom configuration (such as default_inputs, fixed_inputs, or schedule) to LaunchPlan.get_or_create() without providing a name, flytekit 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

Named Launch Plans​

To configure schedules, notifications, or input bindings, supply a unique name to LaunchPlan.get_or_create() or LaunchPlan.create():

from flytekit.core.launch_plan import LaunchPlan

staging_lp = LaunchPlan.get_or_create(
    name="data_pipeline_staging",
    workflow=data_pipeline,
    default_inputs={"dataset": "s3://my-bucket/staging-data"},
    fixed_inputs={"retries": 1},
)

In-Memory Caching and Name Uniqueness​

LaunchPlan maintains an internal dictionary cache (LaunchPlan.CACHE) keyed by the launch plan name (or the workflow name for default launch plans).

  • When you call LaunchPlan.get_or_create() with a name that already exists in CACHE, flytekit checks whether all parameters match the cached instance.
  • If the workflow or any configuration attributes differ from the cached version, flytekit raises an AssertionError to prevent silent misconfiguration.
  • Calling LaunchPlan.create() directly on an existing name will always raise an AssertionError if that name is already in LaunchPlan.CACHE.

Parameterization: Default vs. Fixed Inputs​

Launch plans support two mechanisms for customizing workflow arguments: default_inputs and fixed_inputs.

Workflow Signature: data_pipeline(dataset: str, retries: int)
  ├── default_inputs={"dataset": "s3://staging"}  --> In ParameterMap (caller can override at launch)
  └── fixed_inputs={"retries": 1}                 --> In LiteralMap (stripped from ParameterMap, locked)

Default Inputs​

Inputs defined in default_inputs provide fallback values at launch time. When triggering an execution via the Flyte UI, CLI, or API, users can override these values.

  • Values supplied in default_inputs take precedence over default argument values specified in the Python workflow signature.
  • Internally, LaunchPlan.create() transforms default_inputs into parameter defaults in the launch plan's ParameterMap.

Fixed Inputs​

Inputs defined in fixed_inputs are immutable. Once set on a launch plan, they cannot be modified or overridden when triggering an execution.

  • Internally, flytekit converts fixed_inputs into a LiteralMap using translate_inputs_to_literals().
  • The LaunchPlan.__init__() constructor strips all keys found in fixed_inputs from the public ParameterMap, ensuring callers cannot supply values for those arguments during execution requests.
  • Merged values from default_inputs and fixed_inputs are saved to the launch plan's _saved_inputs dictionary for local executions.
from flytekit.core.launch_plan import LaunchPlan

production_lp = LaunchPlan.get_or_create(
    name="data_pipeline_prod_locked",
    workflow=data_pipeline,
    default_inputs={"dataset": "s3://my-bucket/production-data"},
    fixed_inputs={"retries": 5},
)

Calling Launch Plans Locally and in Workflows​

You can invoke a launch plan directly in Python code, either during local testing or inside another workflow.

Launch plans enforce keyword arguments only. Passing positional arguments raises an AssertionError:

# Valid invocation
result = production_lp(dataset="s3://my-bucket/custom-data")

# Invalid: raises AssertionError("Only Keyword Arguments are supported for launch plan executions")
# result = production_lp("s3://my-bucket/custom-data")

Under the hood, LaunchPlan.__call__() handles two execution modes:

  1. Local Execution (ctx.compilation_state is None): The call merges the launch plan's _saved_inputs (which contains default and fixed inputs) with any explicit keyword arguments passed to __call__(), forwarding execution directly to self.workflow(*args, **inputs).
  2. Compilation / Workflow Embedding (ctx.compilation_state is not None): The call links a sub-launch plan node into the workflow directed acyclic graph (DAG) via create_and_link_node(ctx, entity=self, **inputs).

Scheduling and Automated Triggers​

Flytekit provides schedule constructs that bind directly to launch plans: CronSchedule and FixedRate.

Cron Schedules​

CronSchedule schedules recurring executions using standard 5-field cron expressions or standard cron aliases.

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

@task
def process_partition(timestamp: datetime) -> str:
    return f"Processing window ending at {timestamp.isoformat()}"

@workflow
def hourly_workflow(kickoff_time: datetime) -> str:
    return process_partition(timestamp=kickoff_time)

hourly_lp = LaunchPlan.get_or_create(
    name="hourly_partition_processor",
    workflow=hourly_workflow,
    schedule=CronSchedule(
        schedule="0 * * * *",
        offset="PT15M",
        kickoff_time_input_arg="kickoff_time",
    ),
    auto_activate=True,
)

Key features of CronSchedule:

  • 5-field cron expression or aliases: Pass standard cron syntax (e.g. "0 0 * * *") or aliases (e.g. "@daily", "hourly", "weekly", "monthly"). Flytekit validates these strings with croniter. The legacy AWS 6-field cron_expression parameter is deprecated; passing it raises an AssertionError.
  • Offset parameter: An optional ISO 8601 duration string (such as "PT15M" or "P1D") matched against the regex ([-+]?)P([-+0-9YMWD]+)?(T([-+0-9HMS.,]+)?)?. This shifts the firing time by the specified duration.
  • Kickoff time injection: Set kickoff_time_input_arg to the name of a datetime argument in your workflow signature (e.g. "kickoff_time"). Flyte will inject the scheduled execution timestamp into that argument when the run triggers.

Fixed Rate Intervals​

FixedRate schedules executions at fixed, recurring intervals specified as a datetime.timedelta.

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

interval_lp = LaunchPlan.get_or_create(
    name="heartbeat_every_thirty_minutes",
    workflow=hourly_workflow,
    schedule=FixedRate(
        duration=timedelta(minutes=30),
        kickoff_time_input_arg="kickoff_time",
    ),
)

FixedRate._translate_duration() evaluates the timedelta and converts it to integer quantities of FixedRateUnit.DAY, FixedRateUnit.HOUR, or FixedRateUnit.MINUTE. Resolutions smaller than 1 minute or values with non-zero microseconds raise an AssertionError:

AssertionError: Granularity of less than a minute is not supported for FixedRate schedules.

Triggers and Auto-Activation​

  • trigger vs schedule: Flytekit supports passing a trigger implementing LaunchPlanTriggerBase (such as OnSchedule(schedule=...)) to the trigger parameter.
  • auto_activate: By default, auto_activate=False. Setting auto_activate=True instructs FlyteAdmin to activate the schedule automatically when the launch plan is registered, without requiring an explicit activation command.

Notifications, Security, and Advanced Configuration​

LaunchPlan.create() and LaunchPlan.get_or_create() accept additional metadata and infrastructure policies.

Email, Slack, and PagerDuty Notifications​

Use notification classes from flytekit.core.notification to trigger external alerts upon workflow completion:

from flytekit.core.launch_plan import LaunchPlan
from flytekit.core.notification import Email, PagerDuty, Slack
from flytekit.models.core.execution import WorkflowExecutionPhase

notified_lp = LaunchPlan.get_or_create(
    name="monitored_pipeline_run",
    workflow=data_pipeline,
    default_inputs={"dataset": "s3://my-bucket/live-feed"},
    fixed_inputs={"retries": 3},
    notifications=[
        Email(
            phases=[WorkflowExecutionPhase.FAILED, WorkflowExecutionPhase.TIMED_OUT],
            recipients_email=["ops-team@example.com"],
        ),
        Slack(
            phases=[WorkflowExecutionPhase.SUCCEEDED],
            recipients_email=["alerts-channel@company.slack.com"],
        ),
        PagerDuty(
            phases=[WorkflowExecutionPhase.FAILED],
            recipients_email=["pagerduty-ingest@company.pagerduty.com"],
        ),
    ],
)

Notifications only accept terminal phases: SUCCEEDED, FAILED, ABORTED, and TIMED_OUT. Specifying non-terminal states raises an AssertionError.

Security Context and Identity​

Specify IAM roles or Kubernetes Service Accounts for workflow runs launched by the plan using SecurityContext:

from flytekit.models.security import Identity, SecurityContext

secure_lp = LaunchPlan.get_or_create(
    name="secure_etl_launch_plan",
    workflow=data_pipeline,
    security_context=SecurityContext(
        run_as=Identity(
            iam_role="arn:aws:iam::123456789012:role/DataPipelineExecutionRole",
            k8s_service_account="etl-service-account",
        )
    ),
)

Note on AuthRole: The legacy auth_role parameter is deprecated. If both auth_role and security_context are passed, flytekit raises a ValueError. If only auth_role is passed, flytekit automatically translates it to SecurityContext(run_as=Identity(...)).

Parallelism, Caching, and Output Storage​

Additional runtime settings include:

  • max_parallelism: An integer capping the maximum number of task nodes that can run concurrently within an execution started by this launch plan.
  • overwrite_cache: A boolean flag (True / False) indicating whether executions from this launch plan should force re-execution and overwrite existing task caches.
  • raw_output_data_config: A RawOutputDataConfig(output_dataset_name=...) object setting target cloud storage prefixes for intermediate and raw task data offloading.
from flytekit.models.common import RawOutputDataConfig

configured_lp = LaunchPlan.get_or_create(
    name="custom_configured_pipeline",
    workflow=data_pipeline,
    max_parallelism=10,
    overwrite_cache=True,
    raw_output_data_config=RawOutputDataConfig(output_dataset_name="s3://my-bucket/custom-outputs/"),
)

Dynamic Workflows and Reference Launch Plans​

Dynamic Workflows (node_dependency_hints)​

When a dynamic task (@dynamic) returns sub-launch plan invocations, FlyteAdmin must know about the launch plan entity beforehand. Pass the launch plan object to node_dependency_hints:

from typing import List
from flytekit import dynamic, workflow
from flytekit.core.launch_plan import LaunchPlan

@workflow
def leaf_workflow(x: int) -> int:
    return x * 2

leaf_lp = LaunchPlan.get_or_create(workflow=leaf_workflow, name="leaf_workflow_lp")

@dynamic(node_dependency_hints=[leaf_lp])
def generate_dynamic_runs(count: int) -> List[int]:
    results = []
    for i in range(count):
        results.append(leaf_lp(x=i))
    return results

Reference Launch Plans​

When interacting with a launch plan registered in another Flyte project, domain, or version, use @reference_launch_plan or ReferenceLaunchPlan. This constructs an interface pointer without making network calls to FlyteAdmin during compilation:

from flytekit import workflow
from flytekit.core.launch_plan import reference_launch_plan

@reference_launch_plan(
    project="shared_analytics",
    domain="production",
    name="aggregate_metrics_lp",
    version="v2.1",
)
def aggregate_metrics_lp(dataset: str, retries: int) -> str:
    ...

@workflow
def orchestration_workflow(source_path: str) -> str:
    return aggregate_metrics_lp(dataset=source_path, retries=2)

If the parameter types or names specified in the reference function signature mismatch the remote launch plan's registered interface, FlyteAdmin returns a validation error when registering the outer workflow.