Migrating from AutoMaterializePolicy to AutomationCondition

[OwenKephart]

Background

In Dagster version 1.8.0, AutoMaterializePolicy is marked as deprecated. You can learn more about the reasoning behind this change here.

The functionality provided by this API will be replaced with the AutomationCondition abstraction. This guide covers how to migrate from AutoMaterializePolicy to AutomationCondition.

These APIs are 100% compatible with each other. In fact, all AutoMaterializePolicy objects are automatically converted into AutomationConditions before evaluation. This means that you can migrate assets one by one with no global impact. This guide covers how to migrate individual assets.

Migrating from AutoMaterializePolicy.eager()

If you have an asset defined as:

from dagster import asset, AutoMaterializePolicy

@asset(auto_materialize_policy=AutoMaterializePolicy.eager())
def some_asset(): ...

, then this can be migrated to:

from dagster import asset, AutomationCondition

@asset(automation_condition=AutomationCondition.eager())
def some_asset(): ...

AutomationCondition.eager() is composed of a set of simpler conditions, and has the same purpose as the original AutoMaterializePolicy.eager(). However, it has a few key differences to note:

• For static, dynamic, or multi-partitioned assets, the automation condition will request missing partitions if they became missing after this policy was added to the asset. This prevents situations where adding this policy to an asset with a long history of missing partitions results in an "accidental backfill".
• For time-partitioned assets, the automation condition will only launch materializations against the latest time partition of that asset. In contrast, the auto-materialize policy could launch runs against arbitrarily old historical partitions, with a rate-limit to prevent more than one partition being requested on any given tick.
  • To remove this restriction see here
• To prevent multiple runs being launched for an asset in quick succession, the automation condition will not request work if any parents are currently being materialized, or if the asset itself is currently being materialized. The auto-materialize policy did not have these provisions.
• The auto-materialize policy does a global search for outdated ancestors, and avoids materializing if any are found. This could have highly negative performance characteristics, and often had confusing results, particularly when mapping from partitioned assets to unpartitioned assets, as every single partition of the upstream would need to be in a “good” state in order for any downstreams to materialize.
  • The automation condition does not have such a check. Instead, it will avoid materializing if any parents are currently in progress.

Migrating from AutoMaterializePolicy.lazy()

If you have assets which look like:

from dagster import asset, AutoMaterializePolicy, FreshnessPolicy

@asset(auto_materialize_policy=AutoMaterializePolicy.lazy())
def upstream(): ...

@asset(
    auto_materialize_policy=AutoMaterializePolicy.lazy(),
    freshness_policy=FreshnessPolicy(
        maximum_lag_minutes=30,
        cron_schedule="0 9 * * *",
    ),
)
def downstream(): ...

, then this can be migrated to:

from dagster import asset, AutomationCondition

@asset(automation_condition=AutomationCondition.any_downstream_conditions())
def upstream(): ...

@asset(
    automation_condition=AutomationCondition.on_cron("30 8 * * *"),
)
def downstream(): ...

Note that the upstream asset now has a condition which will be a copy of any downstream condition(s). In this case, it “inherits” the condition of the downstream, meaning it will materialize at 8:30AM every day.

This setup will have nearly identical behavior to the previous version, with a run being kicked off at exactly 8:30AM to refresh both assets. Differences in behavior may occur in cases where there are multiple downstream assets with different cron schedules which do not divide evenly into each other. Previously, the system would attempt to choose a time to execute that landed within the time windows of each independent cron schedule. However, this process was imprecise and frequently led to confusing and unintuitive behavior, and rarely resulted in an optimization that was operationally useful.

We are considering adding utilities to help handle the edge cases of uneven scheduling constraints better, so if you experience undesirable behavior changes after making these updates, let us know and we can work with you to understand your use case.

Migrating from other policies

If you have policies set up with custom rules which cannot be handled by AutomationCondition.on_cron(), then check out the Customizing automation conditions guide, or leave a comment below.

[malow106]

Hi, nice revamp :)

I have some question :

• Is there a plan to see asset automation_condition in the UI like the legacy freshness policies before ?
• How to implement this with +meta on DBT assets
• Is it possible to specify timezone for on_cron() (doc seem broken on this )

Also, assets with freshness_policy will be trigered with the new default_automation_condition_sensor ? Can't make it work

[OwenKephart]

Hi @malow106 !

Is there a plan to see asset automation_condition in the UI

A few angles to this one! First, you can currently see a history of all evaluations of the automation condition on the Asset Details page (under the Automation tab). In terms of being able to visualize the most recent evaluation of the automation condition directly on the asset graph, this is not currently planned, as in most cases an Automation Condition will only be true for a small moment in time (as a run is launched whenever it is true, conditions are generally crafted to become false immediately after a run is requested). In terms of giving better contextual information on an asset node as to what its Automation Condition is (currently you can only find that on the history page), this is planned for the near future.

How to implement this with +meta on DBT assets

We recommend doing this with the DagsterDbtTranslator / the get_automation_condition method. This allows your define an arbitrary function to map from configuration in +meta to an AutomationCondition.

Is it possible to specify timezone for on_cron()

Yes! AutomationCondition.on_cron(cron_schedule="@daily", cron_timezone="US/Pacific") (and a fix for those docs is going out this week)

Also, assets with freshness_policy will be trigered with the new default_automation_condition_sensor ?

AutomationConditions will not respect the freshness_policy defined on an asset. Instead, you should add an explicit cron schedule for that asset which matches the freshness policy. We also recommend using freshness checks instead of freshness policies: https://docs.dagster.io/concepts/assets/asset-checks/checking-for-data-freshness#using-asset-checks-to-check-data-freshness

[malow106]

Thanks a lot, appreciate your detailed feedback 😊

[malow106]

Hi, @OwenKephart, one little question please

When you said :

Previously, the system would attempt to choose a time to execute that landed within the time windows of each independent cron schedule. However, this process was imprecise and frequently led to confusing and unintuitive behavior, and rarely resulted in an optimization that was operationally useful.

Does it mean that today in this scenario AutomationCondition will execute upstream assets multiple times ? (for example : on each distinct on_cron() downstream AutomationCondition)

[OwenKephart]

If those cron schedules are aligned with each other, then a single run can satisfy multiple constraints at once. The simple example would be if you have a downstream hourly schedule and a downstream daily schedule. At midnight, both of those schedules will require the asset to execute, and a single run will accomplish that for both.

[aksestok]

I can't get this example to work:

from dagster import asset, AutomationCondition

@asset(automation_condition=AutomationCondition.any_downstream_conditions())
def upstream(): ...

@asset(
    automation_condition=AutomationCondition.on_cron("30 8 * * *"),
)
def downstream(): ...

What I see happening instead, is that neither the downstream nor the upstream asset ever get materialized.

The downstream asset waits eternally for the upstream assets to get materialized, as the on_cron condition needs all upstream assets to have been updated since the last cron tick:

dagster/python_modules/dagster/dagster/_core/definitions/declarative_automation/automation_condition.py

Line 479 in f4cff85

& AutomationCondition.all_deps_updated_since_cron(cron_schedule, cron_timezone)

See this example evaluation:

Is this as expected, or has something changed since this migration guide was made?

[OwenKephart]

Hi @aksestok -- it looks like in your example, the upstream asset has some further upstream dependencies, which have not been updated.

Running the following example:

from dagster import AutomationCondition, asset


@asset(automation_condition=AutomationCondition.any_downstream_conditions())
def upstream() -> None: ...


@asset(
    automation_condition=AutomationCondition.on_cron("* * * * *"),
    deps=[upstream],
)
def downstream() -> None: ...

results in both assets being executed every minute.

Are those upstream assets external / not possible to materialize or observe? One possibility to ignore those upstreams in the calculation would be to swap out the on_cron condition with a version that ignores the assets you don't expect to update.

This would look like:

def my_on_cron(cron_schedule: str, cron_timezone: str = "UTC") -> "AutomationCondition":
    return (
        AutomationCondition.in_latest_time_window()
        & AutomationCondition.cron_tick_passed(cron_schedule, cron_timezone).since_last_handled()
        & AutomationCondition.all_deps_updated_since_cron(cron_schedule, cron_timezone).ignore(
            # add whatever keys you want to ignore
            selection=AssetSelection.keys(...)
        )
    ).with_label(f"my_on_cron({cron_schedule}, {cron_timezone})")

[aksestok]

Ah, yes, you are right. The direct upstream assets depend on dbt sources that are not observable, which then explains the behaviour I'm seeing. I didn't think of these "stub assets" affecting the evaluation at all. We can just tag these and ignore them in the condition as suggested.

Thanks for the quick reply!

[OwenKephart]

No problem! One additional thing to consider is using AssetSelection.all().upstream_source_assets() as your selection in the case that you never want to wait for a source asset before materializing. The only advantage is that this might be slightly simpler than tagging all the assets one by one, but either way is fine.

[aksestok]

Thanks the tip. In our case, we have a mix of upstream assets where some are observable and some are not, so I think tagging them is one of the easier options for us.