How to: trigger materializing an asset's partition upon upstream asset partition materialization

[clairelin135]

We have a partitioned asset X depends on three other partitioned assets. How can we trigger a run that materializes partition P of asset X whenever partition P has been materialized for the three upstream assets?

[clairelin135]

One possible way to do this is to define a sensor that checks for the existence of materializations for the upstream asset partitions:

from typing import Dict, Set

from dagster import (
    AssetGroup,
    AssetKey,
    AssetMaterialization,
    DagsterEventType,
    DailyPartitionsDefinition,
    EventRecordsFilter,
    Output,
    RunRequest,
    asset,
    job,
    op,
    repository,
    sensor,
)

daily_partitions_def = DailyPartitionsDefinition(start_date="2020-01-01")

@asset(partitions_def=daily_partitions_def)
def table_a(context):
    return 1

@asset(partitions_def=daily_partitions_def)
def table_b(context):
    return 1

@asset(partitions_def=daily_partitions_def)
def table_c(context):
    return 1

@asset(partitions_def=daily_partitions_def)
def dependent_asset(context):
    return 1

asset_group = AssetGroup.from_current_module()
trigger_dependent_asset = asset_group.build_job(name="trigger_dependent_asset", selection=['dependent_asset'])

@sensor(job=trigger_dependent_asset)
def multi_asset_sensor(context):
    partition_keys = daily_partitions_def.get_partitions_keys()
    last_partition_index = int(context.cursor) if context.cursor else -1
    curr_partition = partition_keys[last_partition_index + 1]

    asset_partition_materialized: Dict[AssetKey, bool] = {} # mapping of asset key to dictionary of materialization status by partition

    asset_keys = [AssetKey("table_a"), AssetKey("table_b"), AssetKey("table_c")]
    for asset_key in asset_keys:
        records = context.instance.get_event_records(
            EventRecordsFilter(
                event_type=DagsterEventType.ASSET_MATERIALIZATION,
                asset_key=asset_key,
                asset_partitions=[curr_partition],
            )
        )
        asset_partition_materialized[asset_key] = True if len(records) else False # materialization record exists for partition

    if asset_partition_materialized[AssetKey("table_a")] and asset_partition_materialized[AssetKey("table_b")] and asset_partition_materialized[AssetKey("table_c")]:
        yield trigger_dependent_asset.run_request_for_partition(partition_key=curr_partition, run_key=None)
        context.update_cursor(str(last_partition_index + 1))


@repository
def my_repo():
    return [
        asset_group, trigger_dependent_asset, multi_asset_sensor
    ]

In the above example, because the assets are time-window partitioned, we can use the SensorEvaluationContext.cursor object to track the latest executed partition. Then, we can search only for the materializations of the current partition.

[clairelin135]

Hey @geoHeil, yes that looks correct.

[simonvanderveldt]

@clairelin135 How would you rate the use of context.instance.get_event_records() with a filter on the desired partition vs using context.instance.get_asset_records() and checking the partition value in the returned AssetRecord?
I guess get_event_records() is the better approach since that checks for a specific partition whereas get_asset_records would only allow checking against the latest partition.

[geoHeil]

See the ongoing discussion here: https://dagster.slack.com/archives/C01U954MEER/p1650091358075689?thread_ts=1649909251.959589&cid=C01U954MEER also for me some topics are not really clear yet here.

[clairelin135]

@simonvanderveldt Yes, exactly. get_asset_records would only return the latest materialization for the given asset. get_event_records is what you would be looking for in this case.

[simonvanderveldt]

@clairelin135 All clear, thanks for the confirmation!