Skip to main content

Launch plans, schedules, and fixed inputs

If you have a workflow that trains a model on a specific date with a locked model version and a default environment, you don't want to duplicate the workflow for every parameter combination. A Flyte launch plan lets you layer defaults, fixed values, schedules, and execution options on top of a single workflow definition.

Default launch plans​

Every workflow is automatically associated with a default launch plan. It has no overrides—no default inputs, fixed inputs, schedules, or execution options—and simply uses whatever defaults are declared in the workflow function signature.

Create or retrieve it with LaunchPlan.get_or_create, omitting the name argument:

from flytekit.core.launch_plan import LaunchPlan

@workflow
def my_wf(a: int, c: str) -> str:
    ...

default_lp = LaunchPlan.get_or_create(workflow=my_wf)

LaunchPlan.get_or_create stores the result in a class-level cache keyed by the workflow name (LaunchPlan.CACHE). If you call it twice for the same workflow, you get the same object back. Because the default launch plan is identified by the workflow name, it cannot coexist with a custom launch plan that reuses that exact name.

Named launch plans with defaults and fixed inputs​

When you need to override behavior, give the launch plan a unique name and pass the extra parameters to LaunchPlan.get_or_create. The following example sets a default value for env, locks model_version so it cannot be changed at launch time, and attaches a schedule:

from flytekit.core.launch_plan import LaunchPlan
from flytekit.core.schedule import CronSchedule

@workflow
def training_wf(date: str, model_version: str, env: str = "dev"):
    ...

prod_lp = LaunchPlan.get_or_create(
    name="training_prod",
    workflow=training_wf,
    default_inputs={"env": "prod"},
    fixed_inputs={"model_version": "v1.2"},
    schedule=CronSchedule(schedule="0 3 * * *", kickoff_time_input_arg="date"),
)

LaunchPlan.create is the lower-level factory that get_or_create delegates to, but most user code should call get_or_create because it verifies caching and uniqueness constraints for you.

Default inputs vs. fixed inputs​

default_inputs override the workflow signature defaults but remain launch-time parameters. fixed_inputs are permanently bound and removed from the launch interface.

Inside LaunchPlan.create, the workflow signature is first converted into a ParameterMap. Then the user-supplied default_inputs are merged in with higher precedence:

wf_signature_parameters = transform_inputs_to_parameters(ctx, workflow.python_interface)

temp_inputs = {}
for k, v in default_inputs.items():
    temp_inputs[k] = (workflow.python_interface.inputs[k], v)
temp_interface = Interface(inputs=temp_inputs, outputs={})
temp_signature = transform_inputs_to_parameters(ctx, temp_interface)
wf_signature_parameters._parameters.update(temp_signature.parameters)

Fixed inputs are translated into Flyte literals and stored as a LiteralMap:

fixed_literals = translate_inputs_to_literals(
    ctx,
    incoming_values=fixed_inputs,
    flyte_interface_types=workflow.interface.inputs,
    native_types=workflow.python_interface.inputs,
)
fixed_lm = _literal_models.LiteralMap(literals=fixed_literals)

The stripping happens in LaunchPlan.__init__ (flytekit/core/launch_plan.py):

def _remove_fixed(parameters, fixed_inputs):
    # Ensure fixed inputs are not in parameter map
    return {k: v for k, v in parameters.parameters.items() if k not in fixed_inputs.literals}

Because fixed inputs are removed from parameters, Flyte will not expose them as launch arguments. For convenience, the constructor also stores the merged Python-native values in _saved_inputs so that local execution and serialization can reuse them without reverse-translating from literals every time.

Scheduling workflow runs​

Launch plans support two legacy schedule types and a newer alpha trigger syntax.

CronSchedule​

CronSchedule (flytekit/core/schedule.py) accepts a schedule string, which can be either a cron alias or a standard five-field cron expression:

from flytekit.core.schedule import CronSchedule
from datetime import datetime

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

# Every minute, binding the scheduled time to the workflow input
CronSchedule(
    schedule="*/1 * * * *",
    kickoff_time_input_arg="kickoff_time",
)

Valid aliases include hourly, @hourly, daily, @daily, weekly, @weekly, monthly, @monthly, yearly, and @yearly. Anything else is validated with croniter. The old cron_expression parameter is rejected outright:

if cron_expression:
    raise AssertionError(
        "cron_expression is deprecated and should not be used. Use `schedule` instead. "
        "See the documentation for more information."
    )

FixedRate​

FixedRate accepts a datetime.timedelta and translates it into a day-, hour-, or minute-level fixed-rate schedule:

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

FixedRate(duration=timedelta(minutes=10))

The translator in FixedRate._translate_duration raises an assertion if the granularity is below one minute:

if duration.microseconds != 0 or duration.seconds % _SECONDS_TO_MINUTES != 0:
    raise AssertionError(
        f"Granularity of less than a minute is not supported for FixedRate schedules.  Received: {duration}"
    )

OnSchedule triggers (alpha)​

OnSchedule implements the LaunchPlanTriggerBase protocol and wraps a CronSchedule or FixedRate. You can pass it to the trigger parameter of LaunchPlan.get_or_create instead of the legacy schedule parameter:

from flytekit.core.schedule import OnSchedule

trigger = OnSchedule(CronSchedule(schedule="0 3 * * *"))

OnSchedule.to_flyte_idl simply forwards the wrapped schedule’s protobuf representation.

Execution options​

Beyond inputs and schedules, launch plans carry metadata that is applied to every execution they create:

  • labels / annotations – Labels or Annotations from flytekit.models.common.
  • raw_output_data_config – A RawOutputDataConfig specifying offloaded storage locations.
  • max_parallelism – Caps the number of task nodes that can run in parallel for the entire workflow.
  • security_context – A SecurityContext (e.g., IAM role or Kubernetes service account) for the execution.
  • auth_role – Deprecated. If you supply it, create translates it into a SecurityContext internally, but specifying both auth_role and security_context raises ValueError.
  • overwrite_cache – If True, every execution created by this launch plan will ignore cached results.
  • auto_activate – If True, the launch plan is activated automatically on registration (False by default).

These parameters are stored as properties on the LaunchPlan object and serialized along with it.

Calling a launch plan​

A launch plan is callable. When you invoke it, positional arguments are rejected immediately:

def __call__(self, *args, **kwargs):
    if len(args) > 0:
        raise AssertionError("Only Keyword Arguments are supported for launch plan executions")

During workflow compilation, __call__ merges the saved inputs (defaults + fixed values) with whatever keyword arguments you supplied and delegates to create_and_link_node:

ctx = FlyteContext.current_context()
if ctx.compilation_state is not None:
    inputs = self.saved_inputs
    inputs.update(kwargs)
    return create_and_link_node(ctx, entity=self, **inputs)

During local execution (or when running a dynamic task locally), the same merge happens, but the call is forwarded directly to the underlying workflow:

def _forward_local(self, *args, **kwargs):
    inputs = self.saved_inputs
    inputs.update(kwargs)
    return self.workflow(*args, **inputs)

Because saved_inputs returns a copy of the internal dict (return self._saved_inputs.copy()), repeated calls will not corrupt the stored defaults.

Wiring launch plans into other constructs​

Imperative workflows​

In an imperative workflow, you add a launch plan with add_launch_plan, which is a thin wrapper around add_entity (flytekit/core/workflow.py):

def add_launch_plan(self, launch_plan: _annotated_launch_plan.LaunchPlan, **kwargs) -> Node:
    return self.add_entity(launch_plan, **kwargs)

Both methods call create_node under the hood, so the launch plan is compiled into the DAG just like a task or sub-workflow.

Dynamic tasks​

A dynamic task that wants to return a launch plan must list it in node_dependency_hints. Otherwise the launch plan might not be registered on Flyte Admin before the dynamic task tries to invoke it:

@workflow
def workflow0():
    ...

launchplan0 = LaunchPlan.get_or_create(workflow0)

@dynamic(node_dependency_hints=[launchplan0])
def launch_dynamically():
    # To run a sub-launchplan it must have previously been registered on flyteadmin.
    return [launchplan0] * 10

This pattern is documented in the node_dependency_hints section of flytekit/core/task.py.

ArrayNode mapping​

ArrayNode can map over a LaunchPlan. When it does, it inspects target.fixed_inputs.literals and excludes those inputs from the mapped interface (flytekit/core/array_node.py):

if isinstance(target, (LaunchPlan, FlyteLaunchPlan)) and not isinstance(target, ReferenceLaunchPlan):
    self._excluded_inputs = set(target.fixed_inputs.literals)

This means you can map over a collection for a launch plan’s free parameters while keeping the fixed inputs constant across every mapped invocation.

Gotchas and edge cases​

  • Default launch plans cannot have extras. If you call get_or_create without a name but pass default_inputs, fixed_inputs, schedule, or any other override, you get ValueError: “Only named launchplans can be created that have other properties...”
  • Names must be globally unique. create raises AssertionError if a launch plan with that name already exists in LaunchPlan.CACHE. get_or_create raises AssertionError if a cached launch plan with the same name has different property values.
  • Fixed inputs vanish from the parameter map. Once an input is fixed, it is no longer part of the launch interface. Do not expect to override it at execution time.
  • auth_role is deprecated. Use security_context instead. Supplying both raises ValueError.
  • CronSchedule rejects cron_expression. Use the schedule parameter with a five-field cron or alias.
  • FixedRate does not support sub-minute intervals. A timedelta with microseconds or seconds not divisible by 60 triggers an AssertionError.
  • saved_inputs is returned as a copy. Mutating the dict returned by saved_inputs does not affect the launch plan.
  • All launch plans register themselves automatically. Each LaunchPlan instance appends itself to FlyteEntities.entities in __init__, which the registration pipeline uses to discover what needs to be serialized.
  • ReferenceLaunchPlan is not registered. ReferenceLaunchPlan (flytekit/core/launch_plan.py) acts as a pointer to a launch plan that already exists on the Flyte backend. It performs no network calls; compilation will fail if the interface you declare does not match the remote one.