How can I pass a job run partition_key to another job?

One way to link jobs together like this is to use Asset sensors https://docs.dagster.io/concepts/partitions-schedules-sensors/sensors#asset-sensors

Using this method can I push the partition key to the next job?

Hmm assets would let you trigger the next job but I’m not sure about passing the key. cc @owen,

I believe you can get that off of the upstream job's tags. Inside the asset sensor, this should be something like

asset_event.dagster_event.logging_tags.get("dagster/partition")

thanks @owen, any documentation on this?

not at the moment (it seems), but there should be

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

Something like this might work?

from dagster import AssetKey, asset_sensor

def create_dbt_sensor(key, job):
    @asset_sensor(
        asset_key=AssetKey(key),
        job=job
    )
    def dbt_sensor(context, asset_event):
        yield RunRequest(
            run_key=context.cursor,
            run_config={
                "ops": {
                    "read_materialization": {
                        "config": {
                            "asset_key": asset_event.dagster_event.asset_key.path,
                            "run_date" : asset_event.dagster_event.logging_tags.get("dagster/partition")
                        }
                    }
                }
            },
        )

    return dbt_sensor

hi @Alec Ryan , sorry for the slow response. I think the run config that you want to pass downstream depends on what behavior you want to vary with this partition information. for example, if your downstream job was another dbt job, you might set the resource config for the dbt_cli_resource to set a key in the vars argument to the upstream partition key. if the downstream job is just another partitioned job, you'd probably end up creating that run config via a partitioned config function: https://docs.dagster.io/concepts/partitions-schedules-sensors/partitions#date-partitioned-job

He @owen, no worries. The run config that I want to pass downstream is from a partitioned asset to my dbt assets job. Ideally, that run date that is used in my partition_job

I guess I would frame it more of "grabbing the partition key from the upstream job" than "passing the partition from the upstream job to the sensor", but I might be misunderstanding. Is the basic flow "some partitioned asset job runs" -> have a sensor that detects the updated asset, and based on the partition of the asset that was updated, launch a dbt job with config that will do something for that partition? With that in mind, I now see that it's not really necessary to grab the partition key off the job's tags, as the partition key should also be present on the asset materialization event itself, but wherever you get the partition key from on the event shouldn't make a big difference. Seems to me like there are two parts of the sensor:

@asset_sensor(...)
def my_sensor(context, asset_event):
    asset_partition = ... # either logging tags thing or asset_event.dagster_event.partition, should be identical
    run_config = ... # some way of varying the behavior of your dbt job based on this value
    yield RunRequest(...)

One way of creating some run config that varies the behavior of the dbt job based on the partition is to just set a value in the resource config for the dbt_cli_resource, and another would be to do something like what you show above, where you set some op config values and read them inside the op. was there something about that solution that didn't work? (I'm assuming read_materializations is a custom op you wrote)

Tbh, I haven't had a chance to test that method I shared yet. Read materialization is stolen from the hacker news asset example I think (left as a placeholder) lol

you can configure resources from the run config (example: https://docs.dagster.io/concepts/configuration/config-schema#passing-configuration-to-multiple-ops-in-a-job). you wouldn't need to include the actual dbt_cli_resource -- you could just keep the dbt_cli_resource unconfigured on the job you create, and supply the necessary configuration here.

@owen this is what I'm attempting to run

gotcha -- at a quick glance, this seems like it would work. any issues?

This is the issue I'm running into now @owen https://dagster.slack.com/archives/C01U954MEER/p1650841077638899

oh I see -- because the second job doesn't produce a materialized asset (because the dbt SDA integration doesn't support partitioning at the moment), the materialization from dbt will not have that info available. one way to get this would be something like:

# query the Dagster instance for information about the upstream run
upstream_run = context.instance.get_run_by_id(asset_event.run_id)

# get the run config from that upstream run
upstream_run_config = upstream_run.run_config

# ... parse the run config to figure out the date

I'll admit, it's a bit ugly, but it should do the job

Any thoughts on something like a pipeline config? A variable that can be configured for all downstream assets.

dbt partitioning is definitely something that we want to make possible, but I'd expect it to land more like "a couple months from now" than "a couple weeks from now". and yeah it makes sense to model that process as a software-defined asset (the snowflake table being the produced asset). personally, I'd try to combine the first two jobs into one (which produces two assets), but this pattern seems totally fine as well.

Awesome, good to know. I was considering doing that to be able to visualize the lineage