Skip to main content

Launch plans, schedules, and fixed inputs

Launch plans are the execution boundary for workflows​

When a workflow needs a schedule, launch-time defaults, or values that callers must not change, use a named LaunchPlan. A workflow also gets a cached, association-free default launch plan:

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

LaunchPlan.get_or_create(workflow=my_wf)

LaunchPlan.get_or_create(workflow=...) selects the default plan when name is omitted. LaunchPlan.get_default_launch_plan derives its ParameterMap from workflow.python_interface, copies workflow-signature defaults into the plan's saved inputs, carries workflow default labels and annotations, and creates an empty LiteralMap for fixed inputs. It does not attach schedules, notifications, or other custom associations.

That default is cached by workflow name. Passing any additional association without a name is an error, so use a unique name when you add configuration:

named_plan = LaunchPlan.get_or_create(
    workflow=wf,
    name="scheduled-wf",
    default_inputs={"a": 10},
    fixed_inputs={"c": "production"},
    schedule=CronSchedule(schedule="*/1 * * * *"),
)

The named path is implemented by LaunchPlan.create. It first derives parameters from the workflow interface, then overlays the entries in default_inputs; those values therefore take precedence over defaults declared in the workflow signature. fixed_inputs are converted from Python values to Flyte literals by translate_inputs_to_literals.

Parameters, fixed inputs, and saved inputs​

A launch plan exposes three related views of its inputs:

  • parameters is the launch-time ParameterMap.
  • fixed_inputs is the serialized LiteralMap of values that cannot be changed at launch.
  • saved_inputs is a copy of the original Python values used by local and compilation-time calls.

LaunchPlan.__init__ removes every fixed-input name from the parameter map. In source, the implementation filters the parameter entries before constructing the new ParameterMap:

fixed_parameter_names = set(fixed_inputs.literals)
launch_parameters = {
    key: value
    for key, value in parameters.parameters.items()
    if key not in fixed_parameter_names
}

Consequently, c in the example is not a launch-time parameter. The implementation nevertheless merges both dictionaries into _saved_inputs in create:

default_inputs.update(fixed_inputs)
lp._saved_inputs = default_inputs

A fixed value therefore wins over a same-named default for calls through the plan, while remaining absent from the exposed launch parameters. saved_inputs returns a shallow copy, so updating the returned dictionary does not update the plan's internal map.

Calling a launch plan​

Invoke a launch plan with keyword arguments only. Saved defaults and fixed values form the baseline, and call-site keyword arguments are merged over that baseline:

named_plan(a=20)

LaunchPlan.__call__ rejects positional arguments with an AssertionError. Its behavior then depends on the active FlyteContext:

  • During compilation, it calls create_and_link_node(ctx, entity=self, **inputs), making the launch-plan invocation a node in the compiled graph.
  • Outside compilation, it forwards the merged inputs to the underlying workflow by calling self.workflow(*args, **inputs).

This is why a launch plan can participate as a graph entity while still forwarding to the workflow for local execution. node_creation.create_node accepts LaunchPlan alongside other callable Flyte entities, and FlyteEntities.entities receives each LaunchPlan during construction for later serialization and registration flows.

Scheduling a launch plan​

Attach a native scheduler schedule through the named plan's schedule argument. CronSchedule accepts either a croniter-compatible five-field schedule or an alias such as @daily:

CronSchedule(
    schedule="*/1 * * * *",  # Following schedule runs every min
)

You can also use a fixed-rate interval:

from datetime import timedelta

FixedRate(duration=timedelta(minutes=10))

Pass either object to LaunchPlan.get_or_create or LaunchPlan.create as schedule:

rate_plan = LaunchPlan.get_or_create(
    workflow=wf,
    name="ten-minute-wf",
    schedule=FixedRate(duration=timedelta(minutes=10)),
)

Cron validation and kickoff time​

CronSchedule.__init__ validates the native schedule with _validate_schedule. A value is accepted when it is one of the class's cron aliases— including hourly, daily, weekly, monthly, annually, and their @... forms—or when croniter accepts it. Invalid values raise ValueError.

Use kickoff_time_input_arg when the workflow needs the scheduled kickoff time as an input. The argument is the workflow input name, not a value:

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

schedule = CronSchedule(
    schedule="*/1 * * * *",
    kickoff_time_input_arg="kickoff_time",
)

The schedule docstring notes that the actual timestamp can be a few seconds after the nominal scheduled time. An optional offset is validated by _validate_offset against the implementation's ISO-8601-like duration pattern; invalid offsets raise ValueError.

Do not use the deprecated cron_expression argument. CronSchedule rejects it immediately with an AssertionError and directs callers to schedule=. The old argument represented AWS/CloudWatch-style expressions; its private validator expects six fields and a ? in either the day-of-month or day-of-week position, whereas native schedule= follows the five-field croniter path.

Fixed-rate granularity​

FixedRate._translate_duration represents a timedelta using whole days, hours, or minutes. Exact day durations use the day unit; otherwise exact hour durations use hours; remaining supported durations use minutes. Microseconds and sub-minute remainders raise an AssertionError, so a 30-second interval is not supported.

The trigger form​

The newer trigger syntax is represented by LaunchPlanTriggerBase. OnSchedule adapts either a CronSchedule or FixedRate to that protocol:

trigger = OnSchedule(CronSchedule(schedule="*/1 * * * *"))

triggered_plan = LaunchPlan.get_or_create(
    workflow=wf,
    name="triggered-wf",
    trigger=trigger,
)

OnSchedule.to_flyte_idl() delegates directly to the wrapped schedule's to_flyte_idl(). The schedule and trigger are separate optional fields on LaunchPlan; use the form required by the registration path you are configuring.

Reusing and deriving plans​

Named plans are cached as well. Repeating get_or_create with the same name returns the cached plan only when the workflow and compared configuration values agree. A different workflow or different defaults, fixed inputs, schedule, metadata, security configuration, or cache setting raises an AssertionError. LaunchPlan.create instead raises when the name already exists, so names must be unique in the current registration context.

clone_with creates a new named plan for the same workflow and uses the current plan's properties when replacement values are omitted. It preserves parameters, fixed inputs, schedule, notifications, labels, annotations, raw-output configuration, parallelism, and security context by default. The trigger is not inherited automatically: clone_with passes its trigger argument directly, whose default is None. Most replacement expressions use or, so falsy values cannot reliably clear an existing setting.

Integration with dynamic and mapped execution​

A launch plan is a graph entity rather than only a registration record. If a dynamic task returns a launch plan that must already exist in Flyte Admin, declare it in node_dependency_hints:

@workflow
def workflow0():
    ...

launchplan0 = LaunchPlan.get_or_create(workflow0)

# Specify node_dependency_hints so that launchplan0 will be registered on flyteadmin, despite this being a
# dynamic task.
@dynamic(node_dependency_hints=[launchplan0])
def launch_dynamically():
    # To run a sub-launchplan it must have previously been registered on flyteadmin.
    return [launchplan0] * 10

ArrayNode also accepts launch-plan targets. Its implementation removes a normal launch plan's fixed-input names from the mapped interface and configures the launch-plan array execution mode. The supported launch-plan path is limited to plans with at most one output.

Referencing an already-registered launch plan​

Use ReferenceLaunchPlan when the plan already exists in a Flyte installation and you need a local entity for compilation. Its constructor identifies the remote plan with project, domain, name, and version, and requires the expected input and output types:

reference = ReferenceLaunchPlan(
    project="project",
    domain="dev",
    name="my.launchplan",
    version="abc123",
    inputs={"a": str, "b": int},
    outputs={},
)

ReferenceLaunchPlan does not contact Admin to discover the interface. The supplied interface is used for compilation, so its types must match the registered plan closely enough for compilation and registration checks to succeed. The reference_launch_plan decorator follows the same principle: it transforms the annotated function with transform_function_to_interface(..., is_reference_entity=True) and constructs a ReferenceLaunchPlan from the resulting inputs and outputs.

These choices keep launch-plan behavior explicit: use the cached default for a workflow with no associations, a uniquely named plan for defaults, fixed inputs, and schedules, and a reference plan when the executable has already been registered elsewhere.