How to achieve that dbt assets depends on python assets?

[jordunar]

As long as I have been working on dbt on Dagster I always have faced problems where the dbt assets where loaded the first ones in the DAG and then the python assets ones were loaded. For example, an structure like this:

@dbt_assets(manifest=dbt_manifest_path, select = 'tag:fook') #select='path:models/example/' , select = 'tag:fook'
def dbt_project_test_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()


@asset(compute_kind="python", deps=get_asset_key_for_model([dbt_project_test_dbt_assets], "test_dagster_migration_1"))
def asset_prueba():
    lista = ["a", "b", "c"]
    return Output(
        lista,
        metadata={
            "text_metadata": "longitud lista",
            "lenlista": MetadataValue.json({"len_lista": len(lista)}),
        },
    )

The problem is that I want to do something the other way; I firstly want to create some assets with Python and only then, when these assets are created, execute the dbt models. What I would want to know if there is anything similar as the deps option from the @assets decorators but with the @dbt_assets ones.

Thanks in advanced

[rexledesma]

If you model your upstream dependencies as dbt sources, you can create assets for your dbt sources. These assets will execute before your dbt assets.

See the docs for details: https://docs.dagster.io/integrations/dbt/reference#upstream-dependencies

[jordunar]

Okey, that answer my question but my problem it's not exactly this one. I'm going to explain you my problem: what I am doing right now is migrating to Dagster some processes that were executed before using GCP workflow. Those processes were executed independently (we had one that import some images, another one that used the result of those imports and did operations with them, etc.) as parts of a bigger pipeline. The thing is that we have a dbt project where the models folder is divided in subfolders, and, for each one of the workflows executed we used different subfolders. For example, for workflow 1 we firstly executed a dbt run command on the subfolder 1 and then launched a dataflow job, and for workflow 2 we firstly executed a dbt run command on the subfolder 2 and then launched two different dataflow jobs. This way, as you can see, we had an amount of steps that were executed sequentially and independently.

What I want now is to merge all this workflows in just one DAG using Dagster. Therefore, what I want to achieve is connecting the last step of workflow 1 (which is launching a dataflow job and can be done with a dagster asset) with the first step of workflow 2 (which is the launching of some dbt models). Anyway, this connection is not a connection that looks for the first step of wf 2 uses the information of the last step of wf1; it's just a way of indicating Dagster the order of execution of the things. I have tried to use what you've sent me as a source but, as you can imagine, the last asset is not a table or any source of data (is the launching of a dataflow job) and when I materialize it it fails.

What I want to know if it's possible to do is something like this:

@asset(deps=[launch_test_dataflow], compute_kind="python")
def asset_test1():
    lista = ["a", "b", "c"]
    return Output(
        lista,
        metadata={
            "text_metadata": "longitud lista",
            "lenlista": MetadataValue.json({"len_lista": len(lista)}),
        },
    )


@asset(deps=[asset_test1], compute_kind="python")
def asset_test2():
    lista = ["d", "e", "f"]
    return Output(
        lista,
        metadata={
            "text_metadata": "longitud lista",
            "lenlista": MetadataValue.json({"len_lista": len(lista)}),
        },
    )

, but the second asset, instead of being a python asset, being a dbt asset. As you can see the second asset here does not use the information of the first asset; this deps argument just indicates us that it needs to be executed only when the first one is executed.

Thanks in advanced,

Javier

[bkimjin]

@jordunar were you able to find a solution?

[jordunar]

Yes, but I have to say I have had to look as it has been a while since I did this and maybe Dagster has included something to solve this more easily.

What I did in order to connect a python asset with a dbt asset is selecting on the dagster code the dbt asset using the tag

This is the only thing that is done on dagster code; the rest is done on dbt. Here, on dbt, you have to include the tags in the dbt_project.yml file, specifying to which models folder you are referring to. You also have to include, in the _existing.yaml _ file, the source.

Finally, you have to specify in the first model of the dependent dbt assets (meaning the first as the one the rest depends on, the one/ones on top of hierarchy) a flag of type --depends_on.

[bkimjin]

Gotcha. Based on some feedback from the dagster slack community, this is how I set it up. Within the asset itself, get the name of the asset:

And insert it into the sources.yml as an asset key:

And it looks like this on Dagster:

[jakubczaplicki]

Thanks @bkimjin ! This method worked for me. It looks like it's still a not documented feature.

[leostera]

If it helps, I'm using this translator to automatically stitch the dbt models and the dagster assets:

class DbtDagsterBridgeTranslator(DagsterDbtTranslator):
    def get_asset_key(self, resource_props: Mapping[str, Any]) -> AssetKey:
        """Get asset key for a dbt resource"""
        # For dagster sources, return the asset key without the dagster prefix
        # This makes dagster.vscode_scraper map to just vscode_scraper
        unique_id = resource_props.get("unique_id", "")
        if unique_id.startswith("source.") and resource_props.get("source_name") == "dagster":
            # Return just the table name as the asset key
            return AssetKey([resource_props.get("name")])
        
        # Default behavior for everything else
        return super().get_asset_key(resource_props)

@dbt_assets(
    manifest="../path/to/manifest.json",
    exclude="tag:dagster",  # Exclude all dagster stub sources
    dagster_dbt_translator=DbtDagsterBridgeTranslator()
)
def dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()

Then in my sources.yml I have:

version: 2

sources:
  - name: dagster
    description: "Dagster assets as upstream dependencies"
    tags: ["dagster"]
    tables:
      - name: dagster_asset_name_1
      - name: dagster_asset_name_2
      - name: dagster_asset_name_3

And as long as the table names in sources.yml match the asset names in Dagster, it all works out.

Hope it helps! 🙏🏼