Automating assets using schedules and jobs#

After creating some asset definitions, you may want to automate their materialization.

In this guide, we'll show you one method of accomplishing this by using schedules. To do this for ops, refer to the Automating ops using schedules guide.

By the end of this guide, you'll be able to:

• Create a schedule that directly targets some assets
• Add the new schedule to your project's Definitions object
• Turn the schedule on

Prerequisites#

To follow this guide, you'll need:

• To install Dagster and the Dagster UI. Refer to the Installation guide for more info and instructions.
• Familiarity with:
  • Asset definitions
  • Code locations (Definitions)

Step 1: Define some assets#

The first step in creating a schedule is to define the target assets we want to materialize. Define a few assets in a group named ecommerce_assets:

@asset(group_name="ecommerce_assets")
def orders_asset():
    return 1


@asset(group_name="ecommerce_assets")
def users_asset():
    return 2

Step 2: Define the schedule#

Next, we'll construct the schedule using ScheduleDefinition and use it to target the assets we created in Step 1.

ecommerce_schedule = ScheduleDefinition(
    name="ecommerce_schedule",
    target=AssetSelection.groups("ecommerce_assets"),
    cron_schedule="15 5 * * 1-5",
    default_status=DefaultScheduleStatus.RUNNING,
)

To build the schedule, we:

1. Imported DefaultScheduleStatus and ScheduleDefinition from dagster

2. Created a schedule using ScheduleDefinition that:

   • Targets the assets we defined in Step 1 using AssetSelection.groups
   • Has a cron expression of 15 5 * * 1-5, which translates to Every Monday through Friday of every month at 5:15AM
   • Is turned on by default (default_status). We'll discuss this more in Step 4.

Step 3: Update the Definitions object#

Next, we'll update our project's Definitions object to include the new assets and schedule. This ensures the job and schedule are available to Dagster processes, such as the Dagster UI.

defs = Definitions(
    assets=[orders_asset, users_asset],
    schedules=[ecommerce_schedule],
)

At this point, your code should look like the following:

from dagster import (
    AssetSelection,
    DefaultScheduleStatus,
    Definitions,
    ScheduleDefinition,
    asset,
    define_asset_job,
)


@asset(group_name="ecommerce_assets")
def orders_asset():
    return 1


@asset(group_name="ecommerce_assets")
def users_asset():
    return 2


ecommerce_schedule = ScheduleDefinition(
    name="ecommerce_schedule",
    target=AssetSelection.groups("ecommerce_assets"),
    cron_schedule="15 5 * * 1-5",
    default_status=DefaultScheduleStatus.RUNNING,
)

defs = Definitions(
    assets=[orders_asset, users_asset],
    schedules=[ecommerce_schedule],
)

Step 4: Turn the schedule on#

Schedules must be turned on before they can be used. In our case, we already turned the schedule on by using the default_status parameter in its ScheduleDefinition, but there are a few other ways to do this:

Dagster UICLIPython code

Heads up! Starting or stopping a schedule in the UI will override any default status set in code.

To turn on a schedule in the Dagster UI, navigate to Overview > Schedules:

After the schedule is started, it will begin executing immediately if the dagster-daemon process is running. This process starts automatically when dagster dev is run.

APIs in this guide#