Subchapter 2.14
references/automation/sensors/asset-sensors.mdMarkdown4 KBView on GitHub
Asset sensors monitor asset materializations and trigger jobs when specific assets are materialized.
Use @asset_sensor to monitor a specific asset:
@dg.asset_sensor(asset_key=dg.AssetKey("daily_sales_data"), job=downstream_job)
def sales_data_sensor(context: dg.SensorEvaluationContext, asset_event: dg.EventLogEntry):
# Triggered whenever daily_sales_data is materialized
yield dg.RunRequest(run_key=context.cursor)The sensor is called once per materialization event with the event details in asset_event.
Asset sensors enable dependencies across different jobs:
# Job A contains upstream_asset
@dg.asset
def upstream_asset(): ...
job_a = dg.define_asset_job("job_a", selection=[upstream_asset])
# Job B is triggered when upstream_asset materializes
@dg.asset_sensor(asset_key=dg.AssetKey("upstream_asset"), job=job_b)
def cross_job_sensor(context, asset_event):
return dg.RunRequest()This pattern is useful when you have logically separate jobs that need coordination.
Asset sensors can monitor assets in different code locations:
@dg.asset_sensor(
asset_key=dg.AssetKey("other_location_asset"),
job=my_job,
)
def cross_location_sensor(context, asset_event):
return dg.RunRequest()The sensor can be in a different code location than the monitored asset.
Add conditional logic to control when to trigger:
@dg.asset_sensor(asset_key=dg.AssetKey("daily_sales_data"), job=downstream_job)
def conditional_sensor(context, asset_event):
# Access materialization metadata
metadata = asset_event.dagster_event.event_specific_data.materialization.metadata
row_count = metadata.get("row_count").value if "row_count" in metadata else 0
if row_count > 1000:
return dg.RunRequest()
else:
return dg.SkipReason(f"Row count {row_count} too low, threshold is 1000")Use cases for conditional logic: Trigger downstream processing only when data volume is sufficient, quality checks pass, or specific metadata conditions are met.
Pass runtime configuration to the triggered job:
@dg.asset_sensor(asset_key=dg.AssetKey("source_data"), job=processing_job)
def config_sensor(context, asset_event):
# Extract partition key from the materialization
partition_key = asset_event.dagster_event.logging_tags.get("dagster/partition")
return dg.RunRequest(
run_key=partition_key,
tags={"dagster/partition": partition_key},
)This allows you to propagate partition information or other metadata from the upstream asset to the triggered job.
Use asset sensors when:
Use declarative automation when:
Declarative automation is the recommended approach for asset-centric workflows. Asset sensors remain valuable for triggering actions outside the asset graph or when you need imperative control.