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 that will invoke the workflow every 10 minutes.

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 allows you to specify a schedule using a cron expression:

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:

• 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

Cron aliases

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

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:

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.