# Launch plans
> This bundle contains all pages in the Launch plans section.
> Source: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans ===

# Launch plans

> **📝 Note**
>
> An LLM-optimized bundle of this entire section is available at [`section.md`](https://www.union.ai/docs/v1/union/user-guide/core-concepts/section.md).
> This single file contains all pages in this section, optimized for AI coding agent context.

A launch plan is a template for a workflow invocation.
It brings together:

* A [workflow](https://www.union.ai/docs/v1/union/user-guide/core-concepts/workflows/page.md)
* A (possibly partial) set of inputs required to initiate that workflow
* Optionally, **Core concepts > Launch plans > Notifications** and **Core concepts > Launch plans > Schedules**

When invoked, the launch plan starts the workflow, passing the inputs as parameters.
If the launch plan does not contain the entire set of required workflow inputs, additional input arguments must be provided at execution time.

## Default launch plan

Every workflow automatically comes with a *default launch plan*.
This launch plan does not define any default inputs, so they must all be provided at execution time.
A default launch plan always has the same name as its workflow.

## Launch plans are versioned

Like tasks and workflows, launch plans are versioned.
A launch plan can be updated to change, for example, the set of inputs, the schedule, or the notifications.
Each update creates a new version of the launch plan.

## Custom launch plans

Additional launch plans, other than the default one, can be defined for any workflow.
In general, a given workflow can be associated with multiple launch plans, but a given launch plan is always associated with exactly one workflow.

## Viewing launch plans for a workflow

To view the launch plans for a given workflow, in the UI, navigate to the workflow's page and click **Launch Workflow**.
You can choose which launch plan to use to launch the workflow from the **Launch Plan** dropdown menu.
The default launch plan will be selected by default. If you have not defined any custom launch plans for the workflow, only the default plan will be available.
If you have defined one or more custom launch plans, they will be available in the dropdown menu along with the default launch plan.
For more details, see **Core concepts > Launch plans > Running launch plans**.

## Registering a launch plan

### Registering a launch plan on the command line

In most cases, launch plans are defined alongside the workflows and tasks in your project code and registered as a bundle with the other entities using the CLI (see [Running your code](https://www.union.ai/docs/v1/union/user-guide/development-cycle/running-your-code/page.md)).

### Registering a launch plan in Python with `UnionRemote`

As with all Union.ai command line actions, you can also perform registration of launch plans programmatically with [`UnionRemote`](https://www.union.ai/docs/v1/union/user-guide/development-cycle/union-remote), specifically, `UnionRemote.register_launch_plan`.

### Results of registration

When the code above is registered to Union.ai, it results in the creation of four objects:

* The task `workflows.launch_plan_example.my_task`
* The workflow `workflows.launch_plan_example.my_workflow`
* The default launch plan `workflows.launch_plan_example.my_workflow` (notice that it has the same name as the workflow)
* The custom launch plan `my_workflow_custom_lp` (this is the one we defined in the code above)

### Changing a launch plan

Launch plans are changed by altering their definition in code and re-registering.
When a launch plan with the same project, domain, and name as a preexisting one is re-registered, a new version of that launch plan is created.

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/defining-launch-plans ===

# Defining launch plans

You can define a launch plan with the [`LaunchPlan` class](https://www.union.ai/docs/v1/union/api-reference/flytekit-sdk/packages/flytekit.core.launch_plan).

This is a simple example of defining a launch plan:

```python
import union

@union.workflow
def my_workflow(a: int, b: str) -> str:
    return f"Result: {a} and {b}"

# Create a default launch plan
default_lp = @union.LaunchPlan.get_or_create(workflow=my_workflow)

# Create a named launch plan
named_lp = @union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="my_custom_launch_plan"
)
```

## Default and Fixed Inputs

Default inputs can be overridden at execution time, while fixed inputs cannot be changed.

```python
import union

# Launch plan with default inputs
lp_with_defaults = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="with_defaults",
    default_inputs={"a": 42, "b": "default_value"}
)

# Launch plan with fixed inputs
lp_with_fixed = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="with_fixed",
    fixed_inputs={"a": 100}  # 'a' will always be 100, only 'b' can be specified
)

# Combining default and fixed inputs
lp_combined = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="combined_inputs",
    default_inputs={"b": "default_string"},
    fixed_inputs={"a": 200}
)
```

## Scheduled Execution

```python
import union
from datetime import timedelta
from flytekit.core.schedule import CronSchedule, FixedRate

# Using a cron schedule (runs at 10:00 AM UTC every Monday)
cron_lp = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="weekly_monday",
    default_inputs={"a": 1, "b": "weekly"},
    schedule=CronSchedule(
        schedule="0 10 * * 1",  # Cron expression: minute hour day-of-month month day-of-week
        kickoff_time_input_arg=None
    )
)

# Using a fixed rate schedule (runs every 6 hours)
fixed_rate_lp = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="every_six_hours",
    default_inputs={"a": 1, "b": "periodic"},
    schedule=FixedRate(
        duration=timedelta(hours=6)
    )
)
```

## Labels and Annotations

Labels and annotations help with organization and can be used for filtering or adding metadata.

```python
import union
from flytekit.models.common import Labels, Annotations

# Adding labels and annotations
lp_with_metadata = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="with_metadata",
    default_inputs={"a": 1, "b": "metadata"},
    labels=Labels({"team": "data-science", "env": "staging"}),
    annotations=Annotations({"description": "Launch plan for testing", "owner": "jane.doe"})
)
```

## Execution Parameters

```python
import union

# Setting max parallelism to limit concurrent task execution
lp_with_parallelism = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="with_parallelism",
    default_inputs={"a": 1, "b": "parallel"},
    max_parallelism=10  # Only 10 task nodes can run concurrently
)

# Disable caching for this launch plan's executions
lp_no_cache = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="no_cache",
    default_inputs={"a": 1, "b": "fresh"},
    overwrite_cache=True  # Always execute fresh, ignoring cached results
)

# Auto-activate on registration
lp_auto_activate = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="auto_active",
    default_inputs={"a": 1, "b": "active"},
    auto_activate=True  # Launch plan will be active immediately after registration
)
```

## Security and Authentication

We can also override the auth role (either an iam role or a kubernetes service account) used to execute a launch plan.

```python
import union
from flytekit.models.common import AuthRole
from flytekit import SecurityContext

# Setting auth role for the launch plan
lp_with_auth = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="with_auth",
    default_inputs={"a": 1, "b": "secure"},
    auth_role=AuthRole(
        assumable_iam_role="arn:aws:iam::12345678:role/my-execution-role"
    )
)

# Setting security context
lp_with_security = union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="with_security",
    default_inputs={"a": 1, "b": "context"},
    security_context=SecurityContext(
        run_as=SecurityContext.K8sServiceAccount(name="my-service-account")
    )
)
```

## Raw Output Data Configuration

```python
from flytekit.models.common import RawOutputDataConfig

# Configure where large outputs should be stored
lp_with_output_config = LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="with_output_config",
    default_inputs={"a": 1, "b": "output"},
    raw_output_data_config=RawOutputDataConfig(
        output_location_prefix="s3://my-bucket/workflow-outputs/"
    )
)
```

## Putting It All Together

A pretty comprehensive example follows below. This custom launch plan has d

```python
comprehensive_lp = LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="comprehensive_example",
    default_inputs={"b": "configurable"},
    fixed_inputs={"a": 42},
    schedule=CronSchedule(schedule="0 9 * * *"),  # Daily at 9 AM UTC
    notifications=[
        Notification(
            phases=["SUCCEEDED", "FAILED"],
            email=EmailNotification(recipients_email=["team@example.com"])
        )
    ],
    labels=Labels({"env": "production", "team": "data"}),
    annotations=Annotations({"description": "Daily data processing"}),
    max_parallelism=20,
    overwrite_cache=False,
    auto_activate=True,
    auth_role=AuthRole(assumable_iam_role="arn:aws:iam::12345678:role/workflow-role"),
    raw_output_data_config=RawOutputDataConfig(
        output_location_prefix="s3://results-bucket/daily-run/"
    )
)
```

These examples demonstrate the flexibility of Launch Plans in Flyte, allowing you to customize execution parameters, inputs, schedules, and more to suit your workflow requirements.

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/viewing-launch-plans ===

# Viewing launch plans

## Viewing launch plans in the UI

Select **Launch Plans** in the sidebar to display a list of all the registered launch plans in the project and domain:

![Launch plans list](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/launch-plans/viewing-launch-plans/launch-plans-list.png)

You can search the launch plans by name and filter for only those that are archived.

The columns in the launch plans table are defined as follows:

* **Name**: The name of the launch plan. Click to inspect a specific launch plan in detail.
* **Triggers**:
  * If the launch plan is active, a green **Active** badge is shown. When a launch plan is active, any attached schedule will be in effect and the launch plan will be invoked according to that schedule.
  * Shows whether the launch plan has a [trigger](./reactive-workflows). To filter for only those launch plans with a trigger, check the **Has Triggers** box in the top right.
* **Last Execution**: The last execution timestamp of this launch plan, irrespective of how the last execution was invoked (by schedule, by trigger, or manually).
* **Last 10 Executions**: A visual representation of the last 10 executions of this launch plan, irrespective of how these executions were invoked (by schedule, by trigger, or manually).

Select an entry on the list to go to that specific launch plan:

![Launch plan view](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/launch-plans/viewing-launch-plans/launch-plan-view.png)

Here you can see:
* **Launch Plan Detail (Latest Version)**:
  * **Expected Inputs**: The input and output types for the launch plan.
  * **Fixed Inputs**: If the launch plan includes predefined input values, they are shown here.
* **Launch Plan Versions**: A list of all versions of this launch plan.
* **All executions in the Launch Plan**: A list of all executions of this launch plan.

In the top right you can see if this launch plan is active (and if it is, which version, specifically, is active). There is also a control for changing the active version or deactivating the launch plan entirely.
See [Activating and deactivating](./activating-and-deactivating) for more details.

## Viewing launch plans on the command line with `uctl`

To view all launch plans within a project and domain:

```shell
$ uctl get launchplans \
       --project <project-id> \
       --domain <domain>
```

To view a specific launch plan:

```shell
$ uctl get launchplan \
       --project <project-id> \
       --domain <domain> \
       <launch-plan-name>
```

See the [Uctl CLI](https://www.union.ai/docs/v1/union/api-reference/uctl-cli) for more details.

## Viewing launch plans in Python with `UnionRemote`

Use the method `UnionRemote.client.list_launch_plans_paginated` to get the list of launch plans.

<!-- TODO need to add and link to full UnionRemote documentation to Union docs -- current UnionRemote page does not document all launch plan methods. -->

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/notifications ===

# Notifications

A launch plan may be associated with one or more notifications, which are triggered when the launch plan's workflow is completed.

There are three types of notifications:
* `Email`: Sends an email to the specified recipients.
* `PagerDuty`: Sends a PagerDuty notification to the PagerDuty service (with recipients specified).
  PagerDuty then forwards the notification as per your PagerDuty configuration.
* `Slack`: Sends a Slack notification to the email address of a specified channel. This requires that you configure your Slack account to accept notifications.

Separate notifications can be sent depending on the specific end state of the workflow. The options are:
* `WorkflowExecutionPhase.ABORTED`
* `WorkflowExecutionPhase.FAILED`
* `WorkflowExecutionPhase.SUCCEEDED`
* `WorkflowExecutionPhase.TIMED_OUT`

For example:

```python
from datetime import datetime

import union

from flytekit import (
    WorkflowExecutionPhase,
    Email,
    PagerDuty,
    Slack
)

@union.task
def add_numbers(a: int, b: int, c: int) -> int:
    return a + b + c

@union.task
def generate_message(s: int, kickoff_time: datetime) -> str:
    return f"sum: {s} at {kickoff_time}"

@union.workflow
def my_workflow(a: int, b: int, c: int, kickoff_time: datetime) -> str:
    return generate_message(
        add_numbers(a, b, c),
        kickoff_time,
    )

union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="my_workflow_custom_lp",
    fixed_inputs={"a": 3},
    default_inputs={"b": 4, "c": 5},
    notifications=[
        Email(
            phases=[WorkflowExecutionPhase.FAILED],
            recipients_email=["me@example.com", "you@example.com"],
        ),
        PagerDuty(
            phases=[WorkflowExecutionPhase.SUCCEEDED],
            recipients_email=["myboss@example.com"],
        ),
        Slack(
            phases=[
                WorkflowExecutionPhase.SUCCEEDED,
                WorkflowExecutionPhase.ABORTED,
                WorkflowExecutionPhase.TIMED_OUT,
            ],
            recipients_email=["your_slack_channel_email"],
        ),
    ],
)
```

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/schedules ===

# Schedules

Launch plans let you schedule the invocation of your workflows.
A launch plan can be associated with one or more schedules, where at most one schedule is active at any one time.
If a schedule is activated on a launch plan, the workflow will be invoked automatically by the system at the scheduled time with the inputs provided by the launch plan. Schedules can be either fixed-rate or `cron`-based.

To set up a schedule, you can use the `schedule` parameter of the `LaunchPlan.get_or_create()` method.

## Fixed-rate schedules

In the following example we add a [FixedRate](https://www.union.ai/docs/v1/union/api-reference/flytekit-sdk/packages/flytekit.core.schedule) that will invoke the workflow every 10 minutes.

```python
from datetime import timedelta

import union
from flytekit import FixedRate

@union.task
def my_task(a: int, b: int, c: int) -> int:
    return a + b + c

@union.workflow
def my_workflow(a: int, b: int, c: int) -> int:
    return my_task(a=a, b=b, c=c)

union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="my_workflow_custom_lp",
    fixed_inputs={"a": 3},
    default_inputs={"b": 4, "c": 5},
    schedule=FixedRate(
        duration=timedelta(minutes=10)
    )
)
```
Above, we defined the duration of the `FixedRate` schedule using `minutes`.
Fixed rate schedules can also be defined using `days` or `hours`.

## Cron schedules

A [`CronSchedule`](https://www.union.ai/docs/v1/union/api-reference/flytekit-sdk/packages/flytekit.core.schedule) allows you to specify a schedule using a `cron` expression:

```python
import union
from flytekit import CronSchedule

@union.task
def my_task(a: int, b: int, c: int) -> int:
    return a + b + c

@union.workflow
def my_workflow(a: int, b: int, c: int) -> int:
    return my_task(a=a, b=b, c=c)

union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="my_workflow_custom_lp",
    fixed_inputs={"a": 3},
    default_inputs={"b": 4, "c": 5},
    schedule=CronSchedule(
        schedule="*/10 * * * *"
    )
)
```

### Cron expression format

A `cron` expression is a string that defines a schedule using five space-separated fields, each representing a time unit.
The format of the string is:

```
minute hour day-of-month month day-of-week
```

Each field can contain values and special characters.
The fields are defined as follows:

| Field          | Values              | Special characters |
|----------------|---------------------|--------------------|
| `minute`       | `0-59`              | `* / , -`          |
| `hour`         | `0-23`              | `* / , -`          |
| `day-of-month` | `1-31`              | `* / , - ?`        |
| `month`        | `1-12` or `JAN-DEC` | `* / , -`          |
| `day-of-week`  | `0-6` or` SUN-SAT`  | `* / , - ?`        |

* The `month` and `day-of-week` abbreviations are not case-sensitive.

* The `,` (comma) is used to specify multiple values.
For example, in the `month` field, `JAN,FEB,MAR` means every January, February, and March.

* The `-` (dash) specifies a range of values.
For example, in the `day-of-month` field, `1-15` means every day from `1` through `15` of the specified month.

* The `*` (asterisk) specifies all values of the field.
For example, in the `hour` field, `*` means every hour (on the hour), from `0` to `23`.
You cannot use `*` in both the `day-of-month` and `day-of-week` fields in the same `cron` expression.
If you use it in one, you must use `?` in the other.

* The `/` (slash) specifies increments.
For example, in the `minute` field, `1/10` means every tenth minute, starting from the first minute of the hour (that is, the 11th, 21st, and 31st minute, and so on).

* The `?` (question mark) specifies any value of the field.
For example, in the `day-of-month` field you could enter `7` and, if any day of the week was acceptable, you would enter `?` in the `day-of-week` field.

### Cron expression examples

| Expression         | Description                               |
|--------------------|-------------------------------------------|
| `0 0 * * *`        | Midnight every day.                       |
| `0 12 * * MON-FRI` | Noon every weekday.                       |
| `0 0 1 * *`        | Midnight on the first day of every month. |
| `0 0 * JAN,JUL *`  | Midnight every day in January and July.   |
| `*/5 * * * *`      | Every five minutes.                       |
| `30 2 * * 1`       | At 2:30 AM every Monday.                  |
| `0 0 15 * ?`       | Midnight on the 15th of every month.      |

### Cron aliases

The following aliases are also available.
An alias is used in place of an entire `cron` expression.

| Alias      | Description                                                      | Equivalent to   |
|------------|------------------------------------------------------------------|-----------------|
| `@yearly`  | Once a year at midnight at the start of 1 January.               | `0 0 1 1 *`     |
| `@monthly` | Once a month at midnight at the start of first day of the month. | `0 0 1 * *`     |
| `@weekly`  | Once a week at midnight at the start of Sunday.                  | `0 0 * * 0`     |
| `@daily`   | Once a day at midnight.                                          | `0 0 * * *`     |
| `@hourly`  | Once an hour at the beginning of the hour.                       | `0 * * * *`     |

## kickoff_time_input_arg

Both `FixedRate` and `CronSchedule` can take an optional parameter called `kickoff_time_input_arg`

This parameter is used to specify the name of a workflow input argument.
Each time the system invokes the workflow via this schedule, the time of the invocation will be passed to the workflow through the specified parameter.
For example:

```python
from datetime import datetime, timedelta

import union
from flytekit import FixedRate

@union.task
def my_task(a: int, b: int, c: int) -> int:
    return a + b + c

@union.workflow
def my_workflow(a: int, b: int, c: int, kickoff_time: datetime ) -> str:
    return f"sum: {my_task(a=a, b=b, c=c)} at {kickoff_time}"

union.LaunchPlan.get_or_create(
    workflow=my_workflow,
    name="my_workflow_custom_lp",
    fixed_inputs={"a": 3},
    default_inputs={"b": 4, "c": 5},
    schedule=FixedRate(
        duration=timedelta(minutes=10),
        kickoff_time_input_arg="kickoff_time"
    )
)
```

Here, each time the schedule calls `my_workflow`, the invocation time is passed in the `kickoff_time` argument.

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/activating-and-deactivating ===

# Activating and deactivating

You can set an active/inactive status on launch plans. Specifically:

* Among the versions of a given launch plan (as defined by name), at most one can be set to active.
  All others are inactive.

* If a launch plan version that has a schedule attached is activated, then its schedule also becomes active and its workflow will be invoked automatically according to that schedule.

* When a launch plan version with a schedule is inactive, its schedule is inactive and will not be used to invoke its workflow.

Launch plans that do not have schedules attached can also have an active version.
For such non-scheduled launch plans, this status serves as a flag that can be used to distinguish one version from among the others.
It can, for example, be used by management logic to determine which version of a launch plan to use for new invocations.

Upon registration of a new launch plan, the first version is automatically inactive.
If it has a schedule attached, the schedule is also inactive.
Once activated, a launch plan version remains active even as new, later, versions are registered.

A launch plan version with a schedule attached can be activated through either the UI, `uctl`, or [`UnionRemote`](https://www.union.ai/docs/v1/union/user-guide/user-guide/development-cycle/union-remote).

## Activating and deactivating a launch plan in the UI

To activate a launch plan, go to the launch plan view and click **Add active launch plan** in the top right corner of the screen:

![Activate schedule](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/launch-plans/activating-and-deactivating/add-active-launch-plan.png)

A modal will appear that lets you select which launch plan version to activate:

![Activate schedule](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/launch-plans/activating-and-deactivating/update-active-launch-plan-dialog.png)

This modal will contain all versions of the launch plan that have an attached schedule.
Note that at most one version (and therefore at most one schedule) of a launch plan can be active at any given time.

Selecting the launch plan version and clicking **Update** activates the launch plan version and schedule.
The launch plan version and schedule are now activated. The launch plan will be triggered according to the schedule going forward.

> [!WARNING]
> Non-scheduled launch plans cannot be activated via the UI.
> The UI does not support activating launch plans that do not have schedules attached.
> You can activate them with `uctl` or `UnionRemote`.

To deactivate a launch plan, navigate to a launch plan with an active schedule, click the **...** icon in the top-right corner of the screen beside **Active launch plan**, and click “Deactivate”.

![Deactivate schedule](https://www.union.ai/docs/v1/union/_static/images/user-guide/core-concepts/launch-plans/activating-and-deactivating/deactivate-launch-plan.png)

A confirmation modal will appear, allowing you to deactivate the launch plan and its schedule.

> [!WARNING]
> Non-scheduled launch plans cannot be deactivated via the UI.
> The UI does not support deactivating launch plans that do not have schedules attached.
> You can deactivate them with `uctl` or `UnionRemote`.

## Activating and deactivating a launch plan on the command line with `uctl`

To activate a launch plan version with `uctl`, execute the following command:

```shell
$ uctl update launchplan \
       --activate \
       --project <project-id> \
       --domain <domain> \
       <launch-plan-name> \
       --version <launch-plan-version>
```

To deactivate a launch plan version with `uctl`, execute the following command:

```shell
$ uctl update launchplan \
       --deactivate \
       --project <project-id> \
       --domain <domain> \
       <launch-plan-name> \
       --version <launch-plan-version>
```

See [Uctl CLI](https://www.union.ai/docs/v1/union/api-reference/uctl-cli) for more details.

## Activating and deactivating a launch plan in Python with `UnionRemote`

To activate a launch plan using version `UnionRemote`:

```python
from union.remote import UnionRemote
from flytekit.configuration import Config

remote = UnionRemote(config=Config.auto(), default_project=<project-id>, default_domain=<domain>)
launch_plan = remote.fetch_launch_plan(ame=<launch-plan-name>, version=<launch-plan-version>).id
remote.client.update_launch_plan(launch_plan.id, "ACTIVE")
```

To deactivate a launch plan version using `UnionRemote`:

```python
from union.remote import UnionRemote
from flytekit.remote import Config

remote = UnionRemote(config=Config.auto(), default_project=<project-id>, default_domain=<domain>)
launch_plan = remote.fetch_launch_plan(ame=<launch-plan-name>, version=<launch-plan-version>)
remote.client.update_launch_plan(launch_plan.id, "INACTIVE")
```

<!-- TODO need to add and link to full UnionRemote documentation to Union docs -- current UnionRemote page does not document all launch plan methods. -->

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/running-launch-plans ===

# Running launch plans

## Running a launch plan in the UI

To invoke a launch plan, go to the **Workflows** list, select the desired workflow, click **Launch Workflow**. In the new execution dialog, select the desired launch plan from the **Launch Plan** dropdown menu and click **Launch**.

## Running a launch plan on the command line with `uctl`

To invoke a launch plan via the command line, first generate the execution spec file for the launch plan:

```shell
$ uctl get launchplan \
       --project <project-id>
       --domain <domain> \
       <launch-plan-name> \
       --execFile <execution-spec-file-name>.yaml
```

Then you can execute the launch plan with the following command:

```shell
$ uctl create execution \
       --project <project-id> \
       --domain <domain> \
       --execFile <execution-spec-file-name>.yaml
```

See [Uctl CLI](https://www.union.ai/docs/v1/union/api-reference/uctl-cli) for more details.

## Running a launch plan in Python with `UnionRemote`

The following code executes a launch plan using `UnionRemote`:

```python
import union
from flytekit.remote import Config

remote = union.UnionRemote(config=Config.auto(), default_project=<project-id>, default_domain=<domain>)
launch_plan = remote.fetch_launch_plan(name=<launch-plan-name>, version=<launch-plan-version>)
remote.execute(launch_plan, inputs=<inputs>)
```

See the [UnionRemote](https://www.union.ai/docs/v1/union/user-guide/core-concepts/development-cycle/union-remote) for more details.

## Sub-launch plans

The above invocation examples assume you want to run your launch plan as a top-level entity within your project.
However, you can also invoke a launch plan from *within a workflow*, creating a *sub-launch plan*.
This causes the invoked launch plan to kick off its workflow, passing any parameters specified to that workflow.

This differs from the case of [subworkflows](https://www.union.ai/docs/v1/union/user-guide/core-concepts/workflows/subworkflows-and-sub-launch-plans) where you invoke one workflow function from within another.
A subworkflow becomes part of the execution graph of the parent workflow and shares the same execution ID and context.
On the other hand, when a sub-launch plan is invoked a full, top-level workflow is kicked off with its own execution ID and context.

See [Subworkflows and sub-launch plans](https://www.union.ai/docs/v1/union/user-guide/core-concepts/workflows/subworkflows-and-sub-launch-plans) for more details.

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/reference-launch-plans ===

# Reference launch plans

A reference launch plan references previously defined, serialized, and registered launch plans. You can reference launch plans from other projects and create workflows that use launch plans declared by others.

When you create a reference launch plan, be sure to verify that the workflow interface corresponds to that of the referenced workflow.

> [!NOTE]
> Reference launch plans cannot be run locally. To test locally, mock them out.

## Example

<!-- TODO: Remove the mention of Flytesnacks below -->
In this example, we create a reference launch plan for the [`simple_wf`](https://github.com/flyteorg/flytesnacks/blob/master/examples/basics/basics/workflow.py#L25) workflow from the [Flytesnacks repository](https://github.com/flyteorg/flytesnacks).

1. Clone the Flytesnacks repository:

    ```shell
    $ git clone git@github.com:flyteorg/flytesnacks.git
    ```

2. Navigate to the `basics` directory:

    ```shell
    $ cd flytesnacks/examples/basics
    ```

3. Register the `simple_wf` workflow:

    ```shell
    $ union register --project flytesnacks --domain development --version v1 basics/workflow.py.
    ```

4. Create a file called `simple_wf_ref_lp.py` and copy the following code into it:

    ```python
    import union
    from flytekit import reference_launch_plan

    @reference_launch_plan(
        project="flytesnacks",
        domain="development",
        name="basics.workflow.simple_wf",
        version="v1",
    )

    def simple_wf_lp(
        x: list[int], y: list[int]
    ) -> float:
        return 1.0

    @union.workflow
    def run_simple_wf() -> float:
        x = [-8, 2, 4]
        y = [-2, 4, 7]
        return simple_wf_lp(x=x, y=y)
    ```

5. Register the `run_simple_wf` workflow:

    ```shell
    $ union register simple_wf_ref_lp.py
    ```

6. In the Union.ai UI, run the workflow `run_simple_wf`.

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/mapping-over-launch-plans ===

# Mapping over launch plans

You can map over launch plans the same way you can [map over tasks](https://www.union.ai/docs/v1/union/user-guide/core-concepts/tasks/task-types) to execute workflows in parallel across a series of inputs.

You can either map over a `LaunchPlan` object defined in one of your Python modules or a [reference launch plan](./reference-launch-plans) that points to a previously registered launch plan.

## Launch plan defined in your code

Here we define a workflow called `interest_workflow` that we want to parallelize, along with a launch plan called `interest_workflow_lp`, in a file we'll call `map_interest_wf.py`.
We then write a separate workflow, `map_interest_wf`, that uses a `map` to parallelize `interest_workflow` over a list of inputs.

```python
import union

# Task to calculate monthly interest payment on a loan
@union.task
def calculate_interest(principal: int, rate: float, time: int) -> float:
    return (principal * rate * time) / 12

# Workflow using the calculate_interest task
@union.workflow
def interest_workflow(principal: int, rate: float, time: int) -> float:
    return calculate_interest(principal=principal, rate=rate, time=time)

# Create LaunchPlan for interest_workflow
lp = union.LaunchPlan.get_or_create(
    workflow=interest_workflow,
    name="interest_workflow_lp",
)

# Mapping over the launch plan to calculate interest for multiple loans
@union.workflow
def map_interest_wf() -> list[float]:
    principal = [1000, 5000, 10000]
    rate = [0.05, 0.04, 0.03]  # Different interest rates for each loan
    time = [12, 24, 36]        # Loan periods in months
    return union.map(lp)(principal=principal, rate=rate, time=time)

# Mapping over the launch plan to calculate interest for multiple loans while fixing an input
@union.workflow
def map_interest_fixed_principal_wf() -> list[float]:
    rate = [0.05, 0.04, 0.03]  # Different interest rates for each loan
    time = [12, 24, 36]        # Loan periods in months
    # Note: principal is set to 1000 for all the calculations
    return union.map(lp, bound_inputs={'principal':1000})(rate=rate, time=time)
```

You can run the `map_interest` workflow locally:

```shell
$ union run map_interest_wf.py map_interest_wf
```

You can also run the `map_interest` workflow remotely on Union.ai:

```shell
$ union run --remote map_interest_wf.py map_interest_wf
```

<!-- TODO: Remove up the mention of Flytesnacks below -->
## Previously registered launch plan

To demonstrate the ability to map over previously registered launch plans, in this example, we map over the [`simple_wf`](https://github.com/flyteorg/flytesnacks/blob/master/examples/basics/basics/workflow.py#L25) launch plan from the basic workflow example in the [Flytesnacks repository](https://github.com/flyteorg/flytesnacks).

Recall that when a workflow is registered, an associated launch plan is created automatically. One of these launch plans will be leveraged in this example, though custom launch plans can also be used.

1. Clone the Flytesnacks repository:

    ```shell
    $ git clone git@github.com:flyteorg/flytesnacks.git
    ```

2. Navigate to the `basics` directory:

    ```shell
    $ cd flytesnacks/examples/basics
    ```

3. Register the `simple_wf` workflow:

    ```shell
    $ union register --project flytesnacks --domain development --version v1 basics/workflow.py
    ```

    Note that the `simple_wf` workflow is defined as follows:

    ```python
    @union.workflow
    def simple_wf(x: list[int], y: list[int]) -> float:
        slope_value = slope(x=x, y=y)
        intercept_value = intercept(x=x, y=y, slope=slope_value)
        return intercept_value
    ```

4. Create a file called `map_simple_wf.py` and copy the following code into it:

    ```python
    import union
    from flytekit import reference_launch_plan

    @reference_launch_plan(
        project="flytesnacks",
        domain="development",
        name="basics.workflow.simple_wf",
        version="v1",
    )
    def simple_wf_lp(
        x: list[int], y: list[int]
    ) -> float:
        pass

    @union.workflow
    def map_simple_wf() -> list[float]:
        x = [[-3, 0, 3], [-8, 2, 4], [7, 3, 1]]
        y = [[7, 4, -2], [-2, 4, 7], [3, 6, 4]]
        return union.map(simple_wf_lp)(x=x, y=y)

    ```

    Note the fact that the reference launch plan has an interface that corresponds exactly to the registered `simple_wf` we wish to map over.

5. Register the `map_simple_wf` workflow. Reference launch plans cannot be run locally, so we will register the `map_simple_wf` workflow to Union.ai and run it remotely.

    ```shell
    $ union register map_simple_wf.py
    ```

6. In the Union.ai UI, run the `map_simple_wf` workflow.

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/reactive-workflows ===

# Reactive workflows

Reactive workflows leverage [artifacts](https://www.union.ai/docs/v1/union/user-guide/core-concepts/artifacts) as the medium of exchange between workflows, such that when an upstream workflow emits an artifact, an artifact-driven trigger in a downstream workflow passes the artifact to a new downstream workflow execution.

A trigger is a rule defined in a launch plan that specifies that when a certain event occurs -- for instance, a new version of a particular artifact is materialized -- a particular launch plan will be executed. Triggers allow downstream data consumers, such as machine learning engineers, to automate their workflows to react to the output of upstream data producers, such as data engineers, while maintaining separation of concerns and eliminating the need for staggered schedules and manual executions.

Updating any trigger associated with a launch plan will create a new version of the launch plan, similar to how schedules are handled today. This means that multiple launch plans, each with different triggers, can be created to act on the same underlying workflow. Launch plans with triggers must be activated in order for the trigger to work.

> [!NOTE]
> Currently, there are only artifact event-based triggers, but in the future, triggers will be expanded to include other event-based workflow triggering mechanisms.

## Scope

Since a trigger is part of a launch plan, it is scoped as follows:
* Project
* Domain
* Launch plan name
* Launch plan version

## Trigger types

### Artifact events

An artifact event definition contains the following:
* Exactly one artifact that will activate the trigger when a new version of the artifact is created
* A workflow that is the target of the trigger
* (Optionally) Inputs to the workflow that will be executed by the trigger. It is possible to pass information from the source artifact, the source artifact itself, and other artifacts to the workflow that will be triggered.

For more information, see [Connecting workflows with artifact event triggers](https://www.union.ai/docs/v1/union/user-guide/core-concepts/artifacts/connecting-workflows-with-artifact-event-triggers).

=== PAGE: https://www.union.ai/docs/v1/union/user-guide/core-concepts/launch-plans/concurrency-control ===

# Concurrency control

Concurrency control allows you to limit the number of concurrently running workflow executions for a specific launch plan, identified by its unique `project`, `domain`, and `name`.
This control is applied across all versions of that launch plan.

> [!NOTE]
> To clone and run the example code on this page, see the [Flytesnacks repo](https://github.com/flyteorg/flytesnacks/tree/master/examples/productionizing/).

## How it works

When a new execution for a launch plan with a `ConcurrencyPolicy` is requested, Flyte performs a check to count the number of currently active executions for that same launch plan (`project/domain/name`), irrespective of their versions.

This check is done using a database query that joins the `executions` table with the `launch_plans` table.
It filters for executions that are in an active phase (e.g., `QUEUED`, `RUNNING`, `ABORTING`, etc.) and belong to the launch plan name being triggered.

If the number of active executions is already at or above the `max_concurrency` limit defined in the policy of the launch plan version being triggered, the new execution will be handled according to the specified `behavior`.

## Basic usage

Here's an example of how to define a launch plan with concurrency control:

```python
from flytekit import ConcurrencyPolicy, ConcurrencyLimitBehavior, LaunchPlan, workflow

@workflow
def my_workflow() -> str:
    return "Hello, World!"

# Create a launch plan with concurrency control
concurrency_limited_lp = LaunchPlan.get_or_create(
    name="my_concurrent_lp",
    workflow=my_workflow,
    concurrency=ConcurrencyPolicy(
        max_concurrency=3,
        behavior=ConcurrencyLimitBehavior.SKIP,
    ),
)
```

## Scheduled workflows with concurrency control

Concurrency control is particularly useful for scheduled workflows to prevent overlapping executions:

```python
from flytekit import ConcurrencyPolicy, ConcurrencyLimitBehavior, CronSchedule, LaunchPlan, workflow

@workflow
def scheduled_workflow() -> str:
    # This workflow might take a long time to complete
    return "Processing complete"

# Create a scheduled launch plan with concurrency control
scheduled_lp = LaunchPlan.get_or_create(
    name="my_scheduled_concurrent_lp",
    workflow=scheduled_workflow,
    concurrency=ConcurrencyPolicy(
        max_concurrency=1,  # Only allow one execution at a time
        behavior=ConcurrencyLimitBehavior.SKIP,
    ),
    schedule=CronSchedule(schedule="*/5 * * * *"),  # Runs every 5 minutes
)
```

## Defining the policy

A `ConcurrencyPolicy` is defined with two main parameters:

- `max_concurrency` (integer): The maximum number of workflows that can be running concurrently for this launch plan name.
- `behavior` (enum): What to do when the `max_concurrency` limit is reached. Currently, only `SKIP` is supported, which means new executions will not be created if the limit is hit.

```python
from flytekit import ConcurrencyPolicy, ConcurrencyLimitBehavior

policy = ConcurrencyPolicy(
    max_concurrency=5,
    behavior=ConcurrencyLimitBehavior.SKIP
)
```

## Key behaviors and considerations

### Version-agnostic check, version-specific enforcement

The concurrency check counts all active workflow executions of a given launch plan (`project/domain/name`).
However, the enforcement (i.e., the `max_concurrency` limit and `behavior`) is based on the `ConcurrencyPolicy` defined in the specific version of the launch plan you are trying to launch.

**Example scenario:**

1. Launch plan `MyLP` version `v1` has a `ConcurrencyPolicy` with `max_concurrency = 3`.
2. Three executions of `MyLP` (they could be `v1` or any other version) are currently running.
3. You try to launch `MyLP` version `v2`, which has a `ConcurrencyPolicy` with `max_concurrency = 10`.
   - **Result**: This `v2` execution will launch successfully because its own limit (10) is not breached by the current 3 active executions.
4. Now, with 4 total active executions (3 original + the new `v2`), you try to launch `MyLP` version `v1` again.
   - **Result**: This `v1` execution will **fail**. The check sees 4 active executions, and `v1`'s policy only allows a maximum of 3.

### Concurrency limit on manual trigger

Upon manual trigger of an execution (via `pyflyte` for example) which would breach the concurrency limit, you should see this error in the console:

```bash
_InactiveRpcError:
    <_InactiveRpcError of RPC that terminated with:
        status = StatusCode.RESOURCE_EXHAUSTED
        details = "Concurrency limit (1) reached for launch plan my_workflow_lp. Skipping execution."
    >
```

### Scheduled execution behavior

When the scheduler attempts to trigger an execution and the concurrency limit is met, the creation will fail and the error message from FlyteAdmin will be logged in FlyteScheduler logs.
**This will be transparent to the user. A skipped execution will not appear as skipped in the UI or project execution page**.

## Limitations

### "At most" enforcement

While the system aims to respect `max_concurrency`, it acts as an "at most" limit.
Due to the nature of scheduling, workflow execution durations, and the timing of the concurrency check (at launch time), there might be periods where the number of active executions is below `max_concurrency` even if the system could theoretically run more.

For example, if `max_concurrency` is 5 and all 5 workflows finish before the next scheduled check/trigger, the count will drop.
The system prevents exceeding the limit but doesn't actively try to always maintain `max_concurrency` running instances.

### Notifications for skipped executions

Currently, there is no built-in notification system for skipped executions.
When a scheduled execution is skipped due to concurrency limits, it will be logged in FlyteScheduler but no user notification will be sent.
This is an area for future enhancement.

## Best practices

1. **Use with scheduled workflows**: Concurrency control is most beneficial for scheduled workflows that might take longer than the schedule interval to complete.
2. **Set appropriate limits**: Consider your system resources and the resource requirements of your workflows when setting `max_concurrency`.
3. **Monitor skipped executions**: Regularly check FlyteAdmin logs to monitor if executions are being skipped due to concurrency limits.
4. **Version management**: Be aware that different versions of the same launch plan can have different concurrency policies, but the check is performed across all versions.