Freshness Policy -> Freshness Checks migration guide

[dpeng817]

As of 1.7, the FreshnessPolicy API has been deprecated in favor of two new asset check factories which check the freshness of passed-in assets. The following will serve as a guide to perform the necessary migration to the new APIs, as well as answer questions about why we’re making this change and what it entails.

1. A freshness policy on a time-window partitioned asset.

   If you have a freshness policy specified on a time-window partitioned asset, you should add a freshness check defined by the build_time_partition_freshness_checks API.

   This API functions differently than the original FreshnessPolicy API in a few ways.

   First, a time-window partitioned asset with a freshness policy was considered fresh if ALL completed partitions before the deadline have been materialized, whereas for a time partition freshness check, the asset is considered fresh if the most recent partition before the deadline has arrived, irrespective of earlier partitions.

   Second, the freshness policy API has two different parameters; a cron_schedule, which defines when we should start check for the existence of an asset; and maximum_lag_minutes, which defines the latest partition we should be searching for. For example, a cron_schedule of "0 9 * * *" and a maximum_lag_minutes of 540 would mean that we expect all partitions ending before midnight of the previous day are fresh. In contrast, the freshness check will derive which partition to search for from the cron schedule directly, instead of requiring a lag parameter to also be set. So, simply specifying a cron_schedule of "0 9 * * *" , the check will infer that the partition we expect is the one ending on midnight yesterday, since it’s the earliest partition to finish before 9am today.

   For the given invocation of a FreshnessPolicy, which expects a time-partitioned asset to be fresh by 9am daily:

   @asset(
   	freshness_policy=FreshnessPolicy(cron_schedule="0 9 * * *", max_lag_minutes=540), 
   	partitions_def=DailyPartitionsDefinition(start_date="2021-01-01") 
   )
   def my_asset():
       ...

   You could instead construct an asset and freshness check like so:

   @asset(
   	partitions_def=DailyPartitionsDefinition(
   	  start_date=pendulum.datetime(2020, 1, 1, 0, 0, 0, tz="UTC")
     ) 
   )
   def my_asset():
       ...
       
   freshness_check = build_time_partition_freshness_checks(
     deadline_cron="0 9 * * *",
     assets=[my_asset],
   )

2. A freshness policy on a non-time-partitioned asset.

   We define a non-time-partitioned asset as one that either has no partitions, or has a partitions definition that does not derive from TimeWindowPartitionsDefinition. There are two different parameterizations possible for a non-time-partitioned asset.

   The first is to have a freshness_cron and a maximum_lag_minutes parameter both defined.

   FreshnessPolicy(cron_schedule="0 9 * * *", maximum_lag_minutes=60)

   this is directly analogous to having a freshness check defined by the build_last_update_freshness_checks API, but the lower_bound_delta parameter replaces maximum_lag_minutes, and deadline_cron replaces cron_schedule.

   import datetime
   
   freshness_checks = build_last_update_freshness(deadline_cron="0 9 * * *", lower_bound_delta=datetime.timedelta(minutes=60), assets=...)

   Second, you can have a maximum_lag_minutes parameterization only.

   FreshnessPolicy(maximum_lag_minutes=60)

   which is directly analogous to only setting lower_bound_delta on the freshness check.

   import datetime
   
   freshness_checks = build_last_update_freshness(lower_bound_delta=datetime.timedelta(minutes=60), assets=...)

F.A.Q.

• Why this change? Two reasons:
  • The previous way of computing “data time” for freshness was inflexible and confusing; while the new system allows users to override the data time of an asset more easily.
  • We felt that freshness as it currently exists was too heavyweight in the system. We found that the overdue status was leading to both alert fatigue and confusing automaterialization behavior. In contrast, composing freshness as asset checks allows us to take advantage of all the usability + customizability wins of asset checks (status alerts, custom UI, insights, etc) without needing to introduce a new concept to the system. In general, we want to reduce the number of “concepts” that users need to keep track of in order to have a production-ready dagster deployment, and reintroducing freshness as a type of check felt like an opportunity to do that.
• What happens to the overdue status populated by a freshness policy? The “overdue” status goes away. When an asset is evaluated to be overdue by a freshness check, it shows up as a failed asset check.
• What happens to freshness-based automaterialization? Freshness checks cannot currently be used in automaterialization rules. We are actively working on replacement to this behavior.
• When will I be forced to migrate off of FreshnessPolicy entirely? FreshnessPolicy will be entirely removed by 1.8. We will not remove it before a replacement is landed for freshness-based automaterialization.

[emirkmo]

We will not remove it before a replacement is landed for freshness-based automaterialization.

As we have been using freshness based automation since at least 1.2 (reconciliation sensors), I highly support this part, since automation is one of the two primary benefits of freshness. The second being Freshness based observability, which has been a big component of the benefit of Dagster's UI for our stakeholders. Without seeing the freshness status at a glance nor being able to schedule based off it, it would be a big downgrade.

Mainly, Freshness Policy allows a declarative mapping from a data contract, i.e. some promise about asset freshness written in code and version control, to its scheduling and details on the asset page. So apart from automation, the bigger aspect is solving this translation from data contract to the scheduling and asset overview for data assets, which was quite natural with the freshness policy. (Think about FreshnessPolicy from a grab bag of dbt and python assets from multiple code locations, being translated and effortlessly scheduling themselves, while providing an overview of asset freshness and status at a glass. That is what FreshnessPolicy and automation based on it has easily allowed. This functionality would be excellent to carry forward or improve).

The functionality of Freshness based checks could already be created based on the Freshness Policy, even before the explicit introduction of freshness checks (or even asset checks to be fair). The "check" seem like a side functionality compared to the above uses of automated scheduling and observability.

I would request that the big picture view of freshness as a promise made by the asset be maintained with the new implementation. It should ideally be easy to assign and use this information for scheduling and in the UI, as this the key metric for most assets when building up observability. Checks on data quality are secondary to the data being there or not.

(Completely agree the FreshnessPolicy was opaque and hard to grasp. Glad to see a rework, but worried about wholesale tossing out without having some of the core functionality.)

[OwenKephart]

Hi @emirkmo, thank you for providing this feedback! I’d like to focus on your points regarding the integration of freshness with scheduling, as this is something that (as you note) we’ve been pretty vague about.

TLDR: We are deprecating the current notion of freshness scheduling as it had deep problems. However we are committed to supporting lazy/on-demand scheduling, but will do so in the context a new system. We don’t have exact timing on when the discussed changes will land, but this is the focus of active development.

To frame the issue, I want to break down the current freshness-based scheduling system into the things that work well conceptually, and the things that don’t (from our perspective). At a high level, we want to double down on the useful things and get rid of things that are confusing or actively bad.

What’s good:

• Deferring scheduling decisions to children
  • In many cases, an asset in the middle of the graph does not actually have an opinion on how often it should be executed. It exists to service downstream assets, and so setting an explicit policy (run once a day) is redundant and error prone (as you might later decide that the downstream asset needs to run once every hour, which would require you to manually update all upstream assets with this new frequency).
  • Freshness-based scheduling handles this nicely, as “lazy” assets can adapt to downstream freshness policies
• Connecting expectations about an asset to actions taken on that asset
  • As you bring up in your post, it’s nice to have a contract that can be “self-enforcing”, i.e. the system automatically makes decisions in such a way to abide by the provided limit.

What’s not so good:

• Global, unconfigurable definition of “data time”
  • FreshnessPolicies work off of a global view of the asset graph. Beyond being slow and expensive to calculate, it’s also quite common for this to result in incorrect answers. If a single upstream node in the graph is expected to update just once in awhile, it’s quite challenging to work around this. The AssetCheck based system is more customizable in this regard.
• SLA-based scheduling is hard to define precisely
  • In most cases, there are multiple ways to approach hitting a data freshness SLA, depending on the specific weights you place on minimizing redundant work vs. minimizing risk of missing that SLA. You don’t have any way of precisely predicting exactly how long materializing some set of assets would take, and so the precise time you decide to start your computation is more of an art than a science. This means we need to support some level of customizability at this level, which the existing system does not.
• Understandability / observability / composability
  • The existing freshness-based scheduling system makes decisions in an entirely parallel way from other auto-materialize rules. This means that there is no way to link its outputs (runs) to the complicated system of inputs it used to produce that output decision. This also makes it hard to compose these scheduling decisions with other parts of the system. I.e. you’d struggle to implement something along the lines of “freshness-based scheduling, but with a maximum rate limit”.

Outline of a Plan

As background, we have been moving towards a world in which all scheduling decisions are expressed as an arbitrary boolean expression of conditions, i.e. “(I haven’t been materialized since 8am) & (all of my parents have been materialized since 8am)”. The term for this would be a SchedulingCondition, so you can build up expressions such as:

~SchedulingCondition.updated_since_cron("...")
& SchedulingCondition.all_deps_match(
    SchedulingCondition.updated_since_cron("...")
)

Note: This is a successor to AutoMaterializePolicy, and backwards compatibility will be maintained in the short term. SchedulingConditions can be seen as implementing a superset of the capabilities of AutoMaterializeRules.

The above expression would (roughly) result in an asset being materialized on a given schedule, waiting for its parents to finish before proceeding. For convenience, this sort of “composite condition” would be available through a static constructor, along the lines of SchedulingCondition.on_cron("...").

To tackle the issue of “lazy” scheduling (i.e. assets that only exist for the purpose of satisfying downstream requirements), we plan on introducing a condition that represents a union of all child conditions. This allows you to set an asset’s scheduling condition to something like SchedulingCondition.child_condition(), which would defer all scheduling logic to the downstream asset (and this would work recursively, meaning the asset could look arbitrarily far down the graph to find the “source” policy):

This means that, for any asset with an explicit “freshness policy”, it can be left up to the user to define how that policy would be best satisfied. In most cases, this would probably be a simple cron schedule condition (such as the one described above), which matches the cadence of your freshness policy. To avoid forcing this information to be double-encoded in both the freshness check and the scheduling policy, we can provide a SchedulingCondition.infer_from_freshness_checks() , which would create a condition that is designed to keep the asset in line with its expressed freshness goal. However, the user could at any point customize this policy to better suit their needs.

Once again, we appreciate your feedback, and if you have thoughts on this direction we’re more than happy to hear them! We’d love to talk more with you about this, either on this issue or via Slack / email.