Hey team, I'm looking for clarification on how to create a sensor that takes the output of one asset and uses that output as the input to a job. Say I have two assets (A and B) in two different jobs (A and B) and anytime asset A is materialised, I want job B to start in order to materialise asset B (using asset A's output). However I also want to be able to manually trigger job B on occasion. I currently have this, where am I going wrong? The asset sensor documentation is fairly abstracted so doesn't give the specific example

@asset
def asset_a(config: ObjectClassImUsing):
    <http://logger.info|logger.info>(f'Received input: {config}')
    ...
    return updated_object


@asset
def asset_b(config: ObjectClassImUsing, asset_a: ObjectClassImUsing):
    <http://logger.info|logger.info>(f'Received input: {config}')
    ...
    return updated_more_object


@asset_sensor(asset_key=AssetKey("asset_a"), job=job_b)
def job_b_sensor(context: SensorEvaluationContext, asset_event: EventLogEntry):
    assert asset_event.dagster_event and asset_event.dagster_event.asset_key
    <http://context.log.info|context.log.info>(asset_event)
    yield RunRequest(
        run_key=context.cursor,
        run_config=RunConfig(
            ops={
                "asset_b": "What do I put here to kick off a run of job_b with the output of asset_a?"
            }
        )
    )

I suggest you save the output metadata for asset A (path for filesystem iomanagers is being saved already) so you can load it in your sensor code and pass to the job config. This is if you need to do something special with this matadata. As to loading the actual output of A in B - this will already happen, you can use its value with the asset_a parameter you have in your code. You don’t need to set any special config parameters to id it.

Thank you for the response!! What would passing metadata to sensor code look like? As for loading the actual output, are you saying I could leave the RunRequest blank and asset B would obtain the output of asset A automatically?

Hi Ben, Daniel’s suggestions are good, but it sounds like what you are looking for is the Auto-Materialize Daemon. This will let you set a policy on B to materialize whenever A is materialized.

As for loading the actual output, are you saying I could leave the RunRequest blank and asset B would obtain the output of asset A automatically?

If you are using IO managers, yes. In general you do not need to explicitly pass the dependencies of an asset when it is materialized-- you either let the framework do it for you (IO managers), or you load them in the asset itself.

Thanks for the suggestions guys! I've managed to sort passing the output from

asset_a

to the input of

asset_b

. But now how do I also manually trigger a run of

job_b

by overriding the run config and passing my own values to the input of

asset_b

?

This is a bit of a niche API (not common to use run config to stub upstream values) but you can do this for config (that is launchpad YAML, but can be straightforwardly translated to a

RunConfig

object):

ops:
  asset_b:
    inputs:
      asset_a:
        value: foobar

That doesn't seem to work, do you have the link to the documentation for that API?

hmm possible I screwed something up, let me test locally. Here is docs: https://docs.dagster.io/concepts/io-management/unconnected-inputs#unconnected-inputs

Thank you for the detailed response, it's really handy 😁