Can an asset_sensor trigger a run per materialized partition of an asset? I have a pipeline that dynamically discovers new partitions and emits an AssetMaterialization for each partition to the same asset_key (but with unique partitions). My asset_sensor (which is configured on the same asset_key) only triggers a run for the latest partition. My expectation would be that the asset sensor triggers a run for each asset partition.

Hi Arslan. You can use the multi-asset sensor to achieve this. Something like:

@multi_asset_sensor(asset_keys=[AssetKey("upstream")], job=downstream_job)
def my_sensor(context):
    run_requests_by_partition = {}
    for partition, materialization in context.latest_materialization_records_by_partition(
        AssetKey("upstream")
    ).items():
        run_requests_by_partition[partition] = downstream_job.run_request_for_partition(partition)
        context.advance_cursor({AssetKey("upstream"): materialization})
    return list(run_requests_by_partition.values())

This is exactly what I need. Thank you!

Hi Arslan. You can access the metadata by reading it off of the event log:

event_log_record.event_log_entry.dagster_event.event_specific_data.materialization.metadata_entries

We are currently missing the ability to add run config in

run_request_for_partition

, I can file a PR and hopefully make it out in the next release

Invalid command. Did you mean to create an issue or a discussion? Try

@dagster_bot <issue|docs|discussion> <title>

@Dagster Bot issue provide run config to run_request_for_partition

Created issue at: https://github.com/dagster-io/dagster/issues/9879

Hi @claire I am in a similar situation where I want my asset_sensor to yield a RunRequest when a AssetMaterialization event takes place. My AssetMaterialization is emitted from a op as

context.log_event(
        AssetMaterialization(
            asset_key=AssetKey('my_asset'),
            partition=my_partition_str,
            metadata={
                'my_metadata_key': my_metadata_value
            },
        )
    )

and for the asset_sensor I am implementing your

multi_asset_sensor

suggestion above, where I replace

run_requests_by_partition

with a

RunRequest

as I want to pass a run_config to the job. The AssetMaterialization op and the multi_asset_sensor are in two different repos. My problem: the multi_asset_sensor is failing with the following error

dagster._core.errors.DagsterInvariantViolationError: Cannot get latest materialization by partition for assets with no partitions

Is my approach to pass the partition to the AssetMaterialization incorrect? Many thanks!

As of 1.0.16

run_request_for_partition

now accepts a run config argument!

thanks @claire! Passing the run config via run_request_for_partition now. However, I am still facing the same error

dagster._core.errors.DagsterInvariantViolationError: Cannot get latest materialization by partition for assets with no partitions

raised as soon as I enable the sensor. Is there a way to check whether or not an asset is properly defined with a partition? As mentioned in my prev message, this is how I materialize my asset

context.log_event(
        AssetMaterialization(
            asset_key=AssetKey('my_asset'),
            partition=my_partition_str,
            metadata={
                'my_metadata_key': my_metadata_value
            },
        )
    )

so I d expect my asset to have a partition

ah... I see the issue. So

latest_materialization_by_partition

expects the asset to be a software-defined asset with a partitions definition, so dagster knows the full set of partitions to retrieve. I think what you can use instead is the

context.materialization_records_for_key

method, which should fetch all the unconsumed materializations. You'll be able to fetch the partition off the event record