I know I can create an upstream dependency to a dbt model wi dagster #ask-community

I know I can create an upstream dependency to a db...

Amir Jaber

08/18/2023, 2:21 PM

I know I can create an upstream dependency to a dbt model with

@asset( deps = {AssetKey(["full", "path", "upstream_dbt_model_name"]) })

def *my_custom_asset():*

Is there a way to make my_custom_asset
an upstream dependency of all the dbt models which are downstream of the

upstream_dbt_model_name

David Weber

08/18/2023, 2:47 PM

Hi @Amir Jaber, if I understand you correctly, you can do the following: You have to specify the dagster asset (my_custom_asset) as a source for your dbt model. This is done directly in dbt, in the yaml config (sources config). Here is the documentation that shows you how: https://docs.dagster.io/integrations/dbt/reference#upstream-dependencies

Amir Jaber

08/18/2023, 2:55 PM

Thanks @David Weber, I’ll try this out

David Weber

08/18/2023, 2:55 PM

Sure, let me know if it solves your issue

Amir Jaber

08/21/2023, 8:07 AM

Hi @David Weber I seem to be on the right path, as my custom python asset is now in the right place in the DAG visually. However, when I try to materialise it, I get this error:

dagster._core.errors.DagsterInvalidDefinitionError: Asset [‘dagster’, ‘my_source_name’] specifies a schema with its key prefixes [‘dagster’], but schema my_schema_name was also provided via run config. Schema can only be specified one way.

For context, here’s how my dbt assets are imported:

@dbt_assets(

manifest=manifest_path,

)

def my_dbt_assets(context: OpExecutionContext, dbt: DbtCliResource):

yield from dbt.cli([“build -f”], context=context).stream()

And how my custom asset is initiated

@asset(

deps = {

AssetKey([“integration”, “upstream_dbt_model_name”]),

},

group_name = “python”,

io_manager_key=“io_manager_snowflake”,

key=get_asset_key_for_source([my_dbt_assets], “dagster”),

)

def my_source_name(context):

David Weber

08/21/2023, 8:11 AM

Hi @Amir Jaber, I am definitely not sure, but I think the issue is that you load your source as a Dagster asset. Because you wanted a Dagster asset to be your source model in dbt right? If so, then you don't have to separately load it, as you do in my_source_name But not sure if I am missing something here and if this is connected to the error.

Amir Jaber

08/21/2023, 8:38 AM

Thanks for your answer @David Weber. I was under the understanding from that docs page that I should create a dbt source which references the output table of

my_custom_asset

. Then I make sure my_custom_asset is named the same as that source and is connected to it using this syntax. Is there more nuance here that I might be missing?

David Weber

08/21/2023, 1:04 PM

So if I understood you correctly in your post, you have dagster assets and dbt models. And you want to have the dependency between dagster assets and dbt models - in a sense that dagster assets are upstream (e.g. "before") your dbt models. In that case, you define your last Dagster asset as your dbt source, on which your dbt models depend. You dont have to define the dbt source as a Dagster asset to achieve this. Example: Your last dagster asset:

Copy code

@asset(...)
def my_last_dagster_asset():
  # this asset outputs data, which I want to use as a source in dbt

And then in order to define this as a source, you do the following in dbt in your `sources.yml`:

Copy code

version: 2

sources:
  - name: my_name
    tables:
      - name: my_last_dagster_asset
        meta:
          dagster:
            asset_key: ["my_last_dagster_asset"]

And of course you use this source in your dbt model now:

Copy code

select * from {{ source("dagster", "my_last_dagster_asset") }}

And bam - you have the dependency from your last dagster asset to your first dbt model. There will be no specific "source asset" in dagster, as typically your "source" is the output of your last dagster asset anyway. This is defined here: https://docs.dagster.io/integrations/dbt/reference#upstream-dependencies

Amir Jaber

08/28/2023, 4:07 PM

Thanks @David Weber. I had to figure out some environment variable hackery in order to enable environment separation, but otherwise your write up is exactly what I was looking for!

David Weber

08/28/2023, 4:19 PM

Amazing @Amir Jaber. Happy to hear that!

Open in Slack

Previous Next