The cloud-native orchestrator for the whole development lifecycle, with integrated lineage and observability.

dagster

I found this: <https://github.com/dagster-io/dagster/issues/5699> But the docs for multi_asset_sensors describe different funcitonality.

I updated my asset_sensor to make unique (more unique) run_key and added tags to to the run_request. Appears to function now.

Glad you got it to work--feel free to post again in <#C01U954MEER|dagster-support> if any other issues come up

Thank you for your help. I am having difficulty extracting metadata from asset_materialization in the sensor.  In my case, my asset returns a dict of values on materialization.  In the sensor, I parse those values and pass to Job in RunConfig.

`def ma_multi_download_sensor(context):`
    `# returns a list of materialization event records for a specified asset key.`
    `all_asset_a_events = context.materialization_records_for_key(`
        `asset_key=AssetKey("aid_ma_multi_samp"), limit=5`
    `)`

Currently, the only meta data value I am able to get to is the 'Path' to storage.  The file in storage contains the key value pairs I am looking for.

`for event in all_asset_a_events:`
        `asset_event_metadata = {`
            `key: val.value`
            `for key, val in event.asset_materialization.metadata.items()`
        `}`
        `print(f'{asset_event_metadata}')`

`{'path': '/tmp/tmp3qg0hm9m/storage/aid_ma_multi_samp'}`

Found my issue passing data from asset to sensor.  Needed to set value and metadata in yielded object.