[RFC] Integrating dbt tests with asset checks

[rexledesma]

Important

Dagster asset checks are currently an experimental feature, and the API may change in future releases.

Related discussions: #15880, #16266

---

You can check the following resources for information on the new APIs:

• Loading dbt tests as Dagster asset checks
• Dagster Asset checks

For a previous history of the RFC, see below.

[RFC] Integrating dbt tests with asset checks

Summary

As a follow-up to the experimental release of asset checks in our dagster==1.4.11 release, we have included an experimental implementation of asset checks for dbt tests out of the box.

We'd love to get your feedback after trying these out!

Requirements

1. You must use the new dagster-dbt API's released in dagster-dbt>0.20.0:
   • Your dbt computations should be executed using the DbtCliResource.
   • Your dbt software-defined assets are recommended to be defined using @dbt_assets.
2. You must be on dagster-dbt>=0.20.13 to enable this feature.
3. Because this is an experimental feature, you will need to explicitly opt-in your dbt tests to be modeled as Dagster asset checks. To do this for all of your tests, you can add the following to your dbt_project.yml, then recompile your dbt manifest:

tests:
  +meta:
    dagster:
      asset_check: True

4. After enabling your asset checks with (3), you must invoke and stream events from a dbt command that runs your tests. As a result, you must either invoke dbt test or dbt build in the body of your computation.

Examples

Below, we show correct and incorrect examples of API usage, as well as sample screenshots of the integration on dbt's jaffle_shop.

Correct Examples

# ✅ `dbt test` is invoked after invoking `dbt run`.
# This executes our dbt tests after our dbt models are materialized.
@dbt_assets(manifest=MANIFEST)
def jaffle_shop_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["run"], context=context).stream()
    yield from dbt.cli(["test"], context=context).stream()


# ✅ `dbt build` is invoked. 
# This executes our dbt tests as our dbt models are being materialized.
@dbt_assets(manifest=MANIFEST)
def jaffle_shop_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["build"], context=context).stream()


# ✅ `dbt build` is invoked by default in `load_assets_from_dbt_manifest`. 
# This executes our dbt tests as our dbt models are being materialized.
jaffle_shop_dbt_assets = load_assets_from_dbt_manifest(manifest=MANIFEST)

Incorrect Examples

# ❌ `dbt test` or `dbt build` is not invoked.
@dbt_assets(manifest=dbt_manifest_path)
def jaffle_shop_dbt_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    yield from dbt.cli(["run"], context=context).stream()


# ❌ `dbt build` is not invoked.
jaffle_shop_dbt_assets = load_assets_from_dbt_manifest(manifest=MANIFEST, use_build_command=False)

Screenshots

As an example, here is how the Dagster UI looks with asset checks enabled on dbt's jaffle_shop.

[geoHeil]

I am still running into problems https://dagster.slack.com/archives/C04CW71AGBW/p1696322718554279 for setting this up. @rexledesma

[johannkm]

resolved

[podviaznikov]

I'm also running into the problem.

I did add

    settings=DagsterDbtTranslatorSettings(enable_asset_checks=True)
)```

I materialize the view but I do not see checks UI.

[johannkm]

What type of dbt tests do you have?

[richard-cybersyn]

Would like to add, it's also not populating for us. We have both generic dbt tests as well as custom sql tests.

[rexledesma]

Just to clarify, dbt singular tests are currently not supported as Dagster asset checks out-of-the-box yet. We only ingest dbt generic tests automatically.

We are tracking the feature request for dbt singular tests here: #18595.

[johannkm]

@richard-cybersyn and I spoke offline and we resolved the issue by upgrading dagster from 1.5.6 to 1.5.12. @podviaznikov would you be able to test with the latest dagster version as well?

[marcilj]

Is there a way to make DBT test/asset checks non-blocking?

In all of our current test, the pipeline get halt if a test fail.

[johannkm]

Are you running with dbt build? You could switch to dbt run; dbt test

[marcilj]

I'm running with dbt build yes.
Oh running it with dbt run, dbt test would resolve my issue? Let me try.

Thanks for the quick response.

[marcilj]

Ok so my goal is to create asset checks based out of dbt tests but does not block my pipeline even if they the checks fails.
I will hookup a sensor that trigger and alert if they fail.

I've tried with dbt run and dbt test instead of dbt build but ended up in the same situation where my downstream asset is getting skipped.

def my_dbt_asset(
    context: OpExecutionContext,
    dbt: DbtCliResource,
    config: DbtConfig,
):
    yield from dbt.cli(["build"], context=context).stream()

def my_dbt_asset(
    context: OpExecutionContext,
    dbt: DbtCliResource,
    config: DbtConfig,
):
    yield from dbt.cli(["run"], context=context).stream()
    yield from dbt.cli(["test"], context=context).stream()

Is there anything I'm missing here?

[rexledesma]

You should ensure that your dbt test invocation has raise_on_error=False.

def my_dbt_asset(

    context: OpExecutionContext,

    dbt: DbtCliResource,

    config: DbtConfig,

):

    yield from dbt.cli(["run"], context=context).stream()

    yield from dbt.cli(["test"], context=context, raise_on_error=False).stream()

[CamronBorealis]

This feature appears to rely on using manifest.json schema v9 or above.

I am using dbt-sqlserver, so I am still on dbt-core 1.4.9 with schema v8.