APIs for customizing asset reconciliation / auto-materialization

[sryza]

Motivation

We’ve heard requests for the ability to customize how assets are auto-materialized, on a few different dimensions:

• Whether to materialize assets lazily, based on freshness policies, or eagerly, as soon as upstream changes occur
• Whether or not to auto-materialize assets that are missing
• Whether or not to auto-materialize assets with new code version
• For time-partitioned assets, how far back in time to auto-materialize partitions

Some users want to be able to set different policies on different assets, rather than a blanket policy across all assets.

Proposal

• Deprecate build_asset_reconciliation_sensor.
• Instead, each asset can have an AutoMaterializationPolicy, which determines under what conditions it should be automatically materialized.
• Behind the scenes, the auto-materialization daemon will materialize assets in accordance with their policies.

Code examples

Automatically materialize an asset whenever the assets directly upstream of it are materialized or are observed to have new data:

@asset(
    auto_materialization_policy=AutoMaterializationPolicy(
        on_upstream_changed=True,
    ),
)
def asset1():
    ...

Automatically materialize an asset when necessary in order to meet its freshness policy:

@asset(
    auto_materialization_policy=AutoMaterializationPolicy(
        on_upstream_changed=False,
        for_freshness=True,
    ),
    freshness_policy=FreshnessPolicy(maximum_lag_minutes=60),
)
def asset1():
    ...

Automatically materialize both asset1 and asset2 when necessary in order to meet the freshness policy on asset2:

lazy_policy = AutoMaterializationPolicy(
    on_upstream_changed=False,
    for_freshness=True,
)

@asset(auto_materialization_policy=lazy_policy)
def asset1():
    ...

@asset(
    auto_materialization_policy=lazy_policy,
    freshness_policy=FreshnessPolicy(maximum_lag_minutes=60),
)
def asset2(asset1):
    ...

Apply the same AutoMaterializationPolicy to every asset within a given Python package:

lazy_policy = AutoMaterializationPolicy(
    on_upstream_changed=False,
    for_freshness=True,
)

assets = load_assets_from_package_module(
    my_package_module, auto_materialization_policy=lazy_policy
)

Parameters for AutoMaterializationPolicy

• on_missing: Whether the asset should be materialized if it is missing.
• on_upstream_changed: Whether the should be materialized eagerly after its upstream assets are materialized or are observed to have new data versions.
• for_freshness: Whether the asset should be materialized at whatever time is necessary in order to satisfy downstream freshness policies
• on_code_version_changed: Whether the asset should be materialized whenever its code version doesn’t match the code version used for its most recent materialization.
• min_minutes_between_materializations: This asset should not be materialized more frequently than once per this many minutes.

[plaflamme]

A common case I have is something like this:

@asset()
def asset_a():
  pass

@asset(partitions_def=DailyPartitionsDefinition(start_date="2020-01-01"))
def asset_b():
  pass

@asset(partitions_def=DailyPartitionsDefinition(start_date="2021-01-01"))
def downstream(asset_a, asset_b):
  pass

I want downstream to be recomputed when asset_b changes (adds a partition or otherwise) and because they differ in start_date, I believe AutoMaterializationPolicy would simplify doing this. But the particularity is that asset_a is a Type 2 Slowly Changing Dimension which means that it is not partitioned and it will also sometimes change, but when it does, it does not require the re-computation of downstream. That's because the change to asset_a can only affect future partitions of downstream, not past ones (under normal circumstances).

With AutoMaterializationPolicy(on_upstream_changed=True) on downstream, what happens if asset_a changes? Would it "invalidate" all partitions on downstream? If so, perhaps on_upstream_changed could take a list of AssetKeys as an alternative, i.e.: to indicate which upstream(s) should trigger the policy?

[johannkm]

AutoMaterializePolicy is a few characters shorter but the grammar is a bit suspect

[sryza]

Mmm interesting, I find that to be a compelling alternative

[MarekVigas]

Hey @sryza
Is the on_code_version_changed param still on the roadmap? I would like to use this with dbt models to be able to auto-materialize instantly.

[sryza]

@MarekVigas - we're still interested in adding this. I filed an issue to track: #15686.

We're currently working on a refactor of our auto-materialize implementation to allow more flexible auto-materialize conditions.

cc @OwenKephart

[jonathanneo]

@sryza How do I perform "Automatically materialize both asset1 and asset2 when necessary in order to meet the freshness policy on asset2" in the new API? Is there an example I can take a look at? Thanks!

[sryza]

Setting AutoMaterializePolicy.lazy() on both of those assets should accomplish this.