[RFC] Ergonomic improvements for Dagster’s dbt integration

[rexledesma]

This discussion is now locked, as the new dagster-dbt API's have been released. Thank you to the community for feedback on the early release of these API's.

In our dagster>1.4.0 and dagster-dbt>0.20.0 release, these new API's were unmarked as experimental.

You can check the following resources for information on the new API's:

• Blog post: Orchestrating dbt with Dagster
• Livestream: Supercharge dbt: You might not need dbt Cloud!
• dbt + Dagster tutorial
• dbt + Dagster reference
• dagster-dbt API docs

For a previous history on the RFC, see below.

[RFC] Ergonomic improvements for Dagster’s dbt integration

Summary

This discussion will provide an overview of the API changes to the dagster-dbt integration which will launch as experimental with the integration’s 0.19.6 release, scheduled for May 25, 2023.

We would like to use this document to gather feedback and answer questions about the new APIs. If you have any comments or use cases that are not addressed, please participate in the discussion!

There are three main new interfaces that we plan to release as part of this release:

1. @dbt_assets, a decorator-based interface for users to define software-defined assets from their dbt resources.
2. DbtCli, an improved and lighter interface for invoking dbt commands within the Dagster framework.
3. DbtManifest, a representation of your dbt project’s resources (models, tests, macros, etc), which will be used as arguments when invoking (1) and (2).

Our goal is that this combination of new tools will replace usage of the existing interfaces in the future, as the new tools will provide equivalent and improved functionality, with better ergonomics in composability. Explicitly, the existing interfaces we plan to replace are:

• load_assets_from_dbt_manifest
• load_assets_from_dbt_project
• DbtCliClient

Requirements

• To use the experimental APIs, you must use dagster-dbt>=0.19.6, which is scheduled to be released on May 25, 2023.
• We assume that you are using at least dbt-core>=1.4.0, which was released on January 25, 2023. We include this constraint to make first-class use of dbt’s new consistency in their events and structured logging system.

Notable Changes

Using a decorator to define software-defined assets from dbt

The current dagster-dbt APIs for creating software-defined assets were sufficient for common use cases like invoking dbt run and dbt build on a set of dbt resources. However, as use cases expanded and customization became a first-class requirement, the current interface needs to be more robust to accomplish complex use cases. It is not sustainable to continue adding a function argument to our APIs every time a new use case emerges.

We classified customization concerns into two classes of problems. Included are the function arguments that were used to allow for this behavior. We analyze load_assets_from_dbt_manifest as an example.

1. As a dbt user, I want to customize the metadata about my generated AssetsDefinition.
   1. I want to select certain dbt resources to model as software-defined assets (manifest_json, select, exclude, selected_unique_ids).
   2. I want to customize the name of the underlying computation that materializes my assets (op_name).
   3. I want to customize the name of the asset keys for my dbt resources (key_prefix, source_key_prefix, node_info_to_asset_key).
   4. I want to partition my assets (partitions_def).
   5. I want to assign an arbitrary grouping to my assets (node_info_to_group_fn).
   6. I want to declaratively schedule my assets (node_info_to_freshness_policy_fn, node_info_to_auto_materialize_policy_fn).
   7. I want to customize the metadata properties of my assets to display the Dagster UI (node_info_to_definition_metadata_fn, display_raw_sql).
2. As a dbt user, I want to materialize my AssetsDefinition using an arbitrary compute function.
   1. I want to include arbitrary metadata about the Dagster events emitted from my computation (runtime_metadata_fn).
   2. I want to load assets in downstream computations (io_manager_key).
   3. I want to use dbt run or dbt build (use_build_command).
   4. I want to pass partition information down to my computation (partition_key_to_vars_fn).
   5. I want to modify the name of the dbt resource that is used in my computation (dbt_resource_key).

Our major insight was that the dagster-dbt integration can focus on (1), which is providing utilities for ingesting software-defined assets from dbt. Meanwhile, (2) was essentially giving the user reign about defining their computation, using the Dagster framework, for the software-defined assets they have selected in (1).

By leaning into this insight, we essentially a receive similar interface to what we have for our @asset and @multi_asset decorators in the core Dagster framework! The utilities for (1) essentially give sensible defaults for the arguments to the @asset and @multi_asset decorator. Then, (2) gives the user the freedom within the bounds of the Dagster framework to define their computation.

New Proposed APIs

We introduce three new interfaces to revamp the integration.

1. @dbt_assets, a decorator-based interface for users to define software-defined assets from their dbt resources.
   • This interface takes in a DbtManifest, and dbt selection methods (select, exclude) to choose a set of dbt resources to model as software-defined assets.
   • The user then defines an arbitrary computation to act upon these software-defined assets.
2. DbtCli, an improved and lighter interface for invoking dbt commands within the Dagster framework.
   • This interface essentially emulates the dbt CLI: you pass to it a dbt command, along with any necessary command line flags to modify your dbt configuration or the dbt command.
   • When executing a command, you can optionally hook into the streaming structured dbt logs associated with the in-progress execution.
   • After executing a command, you can optionally retrieve artifacts associated with the dbt command.
3. DbtManifest, a representation of your dbt project’s resources (models, tests, macros, etc), which will be used as arguments when invoking (1) and (2).
   • This is essentially a wrapper around the dbt manifest, and is used as a translation layer between dbt concepts and Dagster concepts.

Examples

We have released an example repository that shows off these new APIs in action against a standard jaffle_shop dbt project.

Later, we will release a dedicated tutorial and additional documentation to explain how to get started with these APIs in an existing dbt project.

Here, we highlight some of the main examples.

1. Invoking dbt run on my entire dbt project

Before 0.19.6

import json

from dagster import file_relative_path
from dagster_dbt import load_assets_from_dbt_manifest

MANIFEST_PATH = file_relative_path(__file__, "../../jaffle_shop/target/manifest.json")

with open(MANIFEST_PATH) as f:
    raw_manifest = json.load(f)

dbt_assets = load_assets_from_dbt_manifest(manifest_json=raw_manifest)

After 0.19.6

import json

from dagster import OpExecutionContext, file_relative_path
from dagster_dbt.asset_decorator import dbt_assets
from dagster_dbt.cli import DbtCli, DbtManifest

MANIFEST_PATH = file_relative_path(__file__, "../../jaffle_shop/target/manifest.json")
manifest = DbtManifest.read(path=MANIFEST_PATH)

@dbt_assets(manifest=manifest)
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    yield from dbt.cli(["run"], manifest=manifest, context=context).stream()

2. Invoking dbt build on my entire dbt project

Before 0.19.6

import json

from dagster import file_relative_path
from dagster_dbt import load_assets_from_dbt_manifest

MANIFEST_PATH = file_relative_path(__file__, "../../jaffle_shop/target/manifest.json")

with open(MANIFEST_PATH) as f:
    raw_manifest = json.load(f)

dbt_assets = load_assets_from_dbt_manifest(
    manifest_json=raw_manifest,
    use_build_command=True,
)

After 0.19.6

import json

from dagster import OpExecutionContext, file_relative_path
from dagster_dbt.asset_decorator import dbt_assets
from dagster_dbt.cli import DbtCli, DbtManifest

MANIFEST_PATH = file_relative_path(__file__, "../../jaffle_shop/target/manifest.json")
manifest = DbtManifest.read(path=MANIFEST_PATH)

@dbt_assets(manifest=manifest)
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    yield from dbt.cli(["build"], manifest=manifest, context=context).stream()

3. Customizing the asset keys for my dbt software-defined assets

Before 0.19.6

import json

from dagster import file_relative_path
from dagster_dbt import load_assets_from_dbt_manifest

MANIFEST_PATH = file_relative_path(__file__, "../../jaffle_shop/target/manifest.json")

with open(MANIFEST_PATH) as f:
   raw_manifest = json.load(f)

dbt_assets = load_assets_from_dbt_manifest(
    manifest_json=raw_manifest,
    key_prefix="prefix",
    source_key_prefix="prefix",
    node_info_to_asset_key=lambda node_info: node_info["alias"],
)

After 0.19.6

import json
from typing import Any, Mapping

from dagster import AssetKey, OpExecutionContext, file_relative_path
from dagster_dbt.asset_decorator import dbt_assets
from dagster_dbt.cli import DbtCli, DbtManifest

MANIFEST_PATH = file_relative_path(__file__, "../../jaffle_shop/target/manifest.json")

class CustomizedDbtManifest(DbtManifest):
    @classmethod
    def node_info_to_asset_key(cls, node_info: Mapping[str, Any]) -> AssetKey:
        return AssetKey(["prefix", node_info["alias"]])

manifest = CustomizedDbtManifest.read(path=MANIFEST_PATH)

@dbt_assets(manifest=manifest)
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    yield from dbt.cli(["run"], manifest=manifest, context=context).stream()

4. Invoking dbt seed, dbt run, dbt test as separate commands

Before 0.19.6

Not possible without forking the load_assets_from_dbt_manifest implementation.

After 0.19.6

import json

from dagster import OpExecutionContext, file_relative_path
from dagster_dbt.asset_decorator import dbt_assets
from dagster_dbt.cli import DbtCli, DbtManifest

MANIFEST_PATH = file_relative_path(__file__, "../../jaffle_shop/target/manifest.json")
manifest = DbtManifest.read(path=MANIFEST_PATH)

@dbt_assets(manifest=manifest)
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    dbt_commands = [
        ["seed"],
        ["run"],
        ["test"],
    ]

    for dbt_command in dbt_commands:
        yield from dbt.cli(dbt_command, manifest=manifest, context=context).stream()

5. Accessing dbt artifacts after invoking your computation

Before 0.19.6

Not possible without forking the load_assets_from_dbt_manifest implementation.

After 0.19.6

import json
import pprint
import sys

from dagster import OpExecutionContext, file_relative_path
from dagster_dbt.asset_decorator import dbt_assets
from dagster_dbt.cli import DbtCli, DbtManifest

MANIFEST_PATH = file_relative_path(__file__, "../../jaffle_shop/target/manifest.json")
manifest = DbtManifest.read(path=MANIFEST_PATH)

@dbt_assets(manifest=manifest)
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    dbt_build = dbt.cli(["build"], manifest=manifest, context=context)

    yield from dbt_build.stream()

    sys.stdout.write(pprint.pformat(dbt_build.get_artifact("run_results.json")))

Next Steps

Again, we would like to use this document to gather feedback and answer questions about the new APIs. If you have any comments or use cases that are not addressed, please participate in the discussion!

[geoHeil]

The @dbt_asset requires the manifest. Does this mean you will no longer support the dynamic generation of the manifest like: load_assets_from_dbt_project?

Where is the io_manager_key specified in the new approach?

In case you want to get pre-created manifests:

1. is the DBT parse good enough to get the desired manifest. Is this even true when macros might require access to the DBT (pivot)?
2. if not now can I get this manifest file without connecting from CI to the prod warehouse instance?

In the past, the target was branch aware. Will this still be true? Or would you like me to take care myself when generating the manifest file?

[geoHeil]

my function looks like:

def my_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    dbt_cli_task = dbt.cli(["run"], manifest=manifest, context=context)

    # Run the task, but don't yield events.
    events = list(dbt_cli_task.stream())

    # Get the run results after the task has completed.
    run_results = dbt_cli_task.get_artifact("run_results.json")
    executed_manifest = dbt_cli_task.get_artifact("manifest.json")
    results_by_output_name = {
        output_name_fn({"unique_id": result["unique_id"]}): result
        for result in run_results["results"]
    }
    manifest_by_output_name = {
        output_name_fn({"unique_id": node["unique_id"]}): node
        for node in executed_manifest["nodes"]
    }

    # Then, we can use the run results to add metadata to the outputs.
    for dagster_event in events:
        if isinstance(dagster_event, Output):
            output_name = dagster_event.output_name
            result = results_by_output_name[output_name]
            rows_affected: Optional[int] = result["adapter_response"].get(
                "rows_affected"
            )
            rows_affected_metadata = (
                {"rows_affected": rows_affected} if rows_affected else {}
            )

            node = manifest_by_output_name[output_name]
            compiled_sql: Optional[str] = node.get("compiled_code")
            compiled_sql_metadata = (
                # TODO what is the right field here to make this show up on the right hand side of dagit?
                {"compiled_sql": compiled_sql} if compiled_sql else {}
            )

            context.add_output_metadata(
                metadata={
                    **rows_affected_metadata,
                    **compiled_sql_metadata,
                },
                output_name=output_name,
            )

        yield dagster_event

Most importantly:

compiled_sql_metadata = (
                # TODO what is the right field here to make this show up on the right hand side of dagit?
                {"compiled_sql": compiled_sql} if compiled_sql else {}
            )

what is the right field here to make it available in dagster?

[geoHeil]

For me, this example code from above is failing with:

FileNotFoundError: [Errno 2] No such file or directory: '
/path/to/dbt/target/dbt_assets-cfb2080-1685635560/run_results.json'

Where is: dbt_assets-cfb2080-1685635560 this not existing folder generated and appended to the path?

[rexledesma]

Got it. Yes, we were not on the same page.

You are talking about the configuration to modify the description of the asset in the Dagster UI. Previously, this was configured in load_assets_from_dbt_project using the display_raw_sql toggle.

You can now accomplish the same thing in @dbt_assets once #14578 lands.

Here's an example:

from typing import Any, Mapping

from dagster import OpExecutionContext
from dagster_dbt.asset_decorator import dbt_assets
from dagster_dbt.cli import DbtCli, DbtManifest

from . import MANIFEST_PATH


class CustomizedDbtManifest(DbtManifest):
    @classmethod
    def node_info_to_description(cls, node_info: Mapping[str, Any]) -> str:
        description_sections = []

        description = node_info.get("description", "")
        if description:
            description_sections.append(description)

        compiled_code = node_info.get("compiled_code", "")
        if compiled_code:
            description_sections.append(f"#### Compiled SQL:\n```\n{compiled_code}\n```")

        return "\n\n".join(description_sections)


manifest = CustomizedDbtManifest.read(path=MANIFEST_PATH)


@dbt_assets(manifest=manifest)
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    yield from dbt.cli(["build"], manifest=manifest, context=context).stream()

[geoHeil]

I think we are still not on the same page: You assume that the manifest contains the compiled_code. And I would need to yield an asset materialization event for the compiled_code during runtime. Does this mean I need to emit an asset materialization event for display_raw_sql: compiled_code to get it to display?

Secondly: #14477 (reply in thread)
Only: /path/to/dbt is set as the path to the DBt project dir usually ./target/run_results.json should be used then. However, dagster is generating: /path/to/dbt/target/dbt_assets-cfb2080-1685635560/run_results.json' and appending dbt_assets-cfb2080-1685635560 which then results in a file not found error. when trying to retrieve the run results.

Does this events = list(dbt_cli_task.stream()) somehow need to be blocking instead for the files to be generated and available?

[rexledesma]

We discussed this offline. I will summarize our conversation here:

You assume that the manifest contains the compiled_code. And I would need to yield an asset materialization event for the compiled_code during runtime. Does this mean I need to emit an asset materialization event for display_raw_sql: compiled_code to get it to display?

The view in question is the sidebar for the AssetsDefinition.

This is populated by the description field of the AssetOut that is generated for you in @dbt_assets. This is not generated at runtime. This is generated at definition time before you even materialize your software-defined asset. To customize this, #14578 will be the solution in the future.

The compiled code must be present at definition time to have it in this menu, so using dbt compile to generate the manifest is required if you want that information.

Secondly: #14477 (reply in thread)
Only: /path/to/dbt is set as the path to the DBt project dir usually ./target/run_results.json should be used then. However, dagster is generating: /path/to/dbt/target/dbt_assets-cfb2080-1685635560/run_results.json' and appending dbt_assets-cfb2080-1685635560 which then results in a file not found error. when trying to retrieve the run results.

This was due to a configuration error, as your postgres database was not set up. This prevent the dbt command from even producing a run_results.json. However, we should have surfaced this configuration error for the user. We will resolve this by raising failures on non-zero exit codes.

Does this events = list(dbt_cli_task.stream()) somehow need to be blocking instead for the files to be generated and available?

list(dbt_cli_task.stream()) is already blocking, and waits for the dbt command to terminate.

[geoHeil]

Previously I had a function like:

@op(required_resource_keys={"dbt"})
def generate_dbt_metadata(context):
    dbt_resource: DbtCliResource = context.resources.dbt
    docs = dbt_resource.generate_docs(compile_project=False)
    tests = dbt_resource.test()
    return docs, tests

which ran outside of the normal dbt run schedule to generate the documentation and compute the test cases and push the metadata to my data catalog.

Now, I see that the @asset notation should be used. But I would not intend to publish metadata assets to dagsters graph. How should this be migrated to the new API?

[rexledesma]

To be clear, we are not mandating that @asset notation be used. This discussion is about ergonomic improvements for generating assets out-of-the-box for our integration, which already exists in the form of load_assets_from_dbt_manifest and load_assets_from_dbt_project.

If you're still using @op's for other computations with dbt, we still support that! You can accomplish that using the DbtCli, which is just a lightweight alternative to the existing DbtCliResource: rather than needing to translate dbt commands to python arguments and Dagster configuration, you can run commands directly.

In your case, you could do something like the following (this uses the new Pythonic resources update, but you could still use required_resource_keys if you want):

@op
def generate_dbt_metadata(dbt: DbtCli):
    docs_events = dbt.cli(["docs", "generate", "--no-compile"], manifest=manifest).wait()
    test_events = dbt.cli(["test"], manifest=manifest).wait()
    
    return docs_events, test_events

[geoHeil]

great

[sungchun12]

Have you folks considered using the new interfaces in 1.5?

Similar in mechanics to what you have built but native to dbt. Disclaimer: still early days, so it's subject to change.

https://docs.getdbt.com/reference/programmatic-invocations

Simple run example

from dbt.cli.main import dbtRunner, dbtRunnerResult

# initialize
dbt = dbtRunner()

# create CLI args as a list of strings
cli_args = ["run", "--select", "tag:my_tag"]

# run the command
res: dbtRunnerResult = dbt.invoke(cli_args)

# inspect the results
for r in res.result:
    print(f"{r.node.name}: {r.status}")

Do fancier things with the manifest.json example

from dbt.cli.main import dbtRunner, dbtRunnerResult
from dbt.contracts.graph.manifest import Manifest

# use 'parse' command to load a Manifest
res: dbtRunnerResult = dbtRunner().invoke(["parse"])
manifest: Manifest = res.result

# introspect manifest
# e.g. assert every public model has a description
for node in manifest.nodes.values():
    if node.resource_type == "model" and node.access == "public":
        assert node.description != "", f"{node.name} is missing a description"

# reuse this manifest in subsequent commands to skip parsing
dbt = dbtRunner(manifest=manifest)
cli_args = ["run", "--select", "tag:my_tag"]
res = dbt.invoke(cli_args)

[rexledesma]

Yes, we've considered it! We took some inspiration from the APIs released in 1.5. However, as you pointed out, the APIs still need to be contracted, which is why we're hesitant to move the underlying implementation to it. Once they've matured, and there's documentation and stability around the objects that are returned by dbtRunner, we can definitely consider moving over.

[dennisgera]

I am currently using the load_assets_from_dbt_manifest to load all my assets once and then declare multiple DbtManifestAssetSelection to select dbt models that I want to run in dagster jobs. If I switch to @dbt_assets, can I continue using DbtManifestAssetSelection to create asset selections of my dbt assets, or should I take a different approach altogether?

This is an example of my current configuration:

dbt_assets = load_assets_from_dbt_manifest(
    manifest_file,
    io_manager_key="io_manager",
    key_prefix=["analytics"],
    source_key_prefix=["analytics"],
    node_info_to_group_fn=node_info_to_group_fn,
)

antifraud_dbt_assets = DbtManifestAssetSelection(
    manifest_json=manifest_file,
    select="tag:antifraud",
    node_info_to_asset_key=node_info_to_asset_key,
)

antifraud_assets_job = define_asset_job(
    name="antifraud_assets_job",
    executor_def=in_process_executor,
    selection=antifraud_dbt_assets,
)

I then declare antifraud_assets_job in my repository.

[rexledesma]

Hi @dennisgera!

If you want to use @dbt_assets along with asset selections, you have a couple of options:

1. Assume that the compute function for your dbt assets is the same. You can define one set of AssetsDefinition using @dbt_assets, and then use DbtManifestAssetSelection, as you've been doing before. Here's roughly what that looks like with your example.

from typing import Any, Mapping

from dagster import AssetKey, OpExecutionContext, define_asset_job
from dagster_dbt import DbtManifestAssetSelection
from dagster_dbt.asset_decorator import dbt_assets
from dagster_dbt.cli import DbtCli, DbtManifest

from . import MANIFEST_PATH


class CustomizedDbtManifest(DbtManifest):
    @classmethod
    def node_info_to_asset_key(cls, node_info: Mapping[str, Any]) -> AssetKey:
        return AssetKey(["analytics", node_info["alias"]])


manifest = CustomizedDbtManifest.read(path=MANIFEST_PATH)


@dbt_assets(manifest=manifest)
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    yield from dbt.cli(["build"], manifest=manifest, context=context).stream()


staging_dbt_assets = DbtManifestAssetSelection(
    manifest_json=manifest.raw_manifest,
    select="fqn:staging.*",
    node_info_to_asset_key=CustomizedDbtManifest.node_info_to_asset_key,
)

staging_assets_job = define_asset_job(
    name="staging_assets_job",
    selection=staging_dbt_assets,
)

2. Assume that the compute function for your dbt assets is different. For example, you may have a case where you'd like to process certain sets of dbt assets in a customized way. In this case, you will need to specify the compute function for each of these sets of assets. @dbt_assets will let you do this, which was not possible with load_assets_from_dbt_manifest.

from typing import Any, Mapping

from dagster import AssetKey, OpExecutionContext, define_asset_job
from dagster_dbt import DbtManifestAssetSelection
from dagster_dbt.asset_decorator import dbt_assets
from dagster_dbt.cli import DbtCli, DbtManifest

from . import MANIFEST_PATH


class CustomizedDbtManifest(DbtManifest):
    @classmethod
    def node_info_to_asset_key(cls, node_info: Mapping[str, Any]) -> AssetKey:
        return AssetKey(["analytics", node_info["alias"]])


manifest = CustomizedDbtManifest.read(path=MANIFEST_PATH)


@dbt_assets(manifest=manifest, exclude="fqn:staging.*")
def my_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    yield from dbt.cli(["build"], manifest=manifest, context=context).stream()


@dbt_assets(manifest=manifest, select="fqn:staging.*")
def my_staging_dbt_assets(context: OpExecutionContext, dbt: DbtCli):
    context.log.info("Start customized behavior here.")

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

    context.log.info("End customized behavior here.")


staging_dbt_assets = DbtManifestAssetSelection(
    manifest_json=manifest.raw_manifest,
    select="fqn:staging.*",
    node_info_to_asset_key=CustomizedDbtManifest.node_info_to_asset_key,
)

staging_assets_job = define_asset_job(
    name="staging_assets_job",
    selection=staging_dbt_assets,
)

[yusufshalaby]

Big fan of these changes, feel like I have a lot of control over my dbt run configs now :)

It would be great if the DbtCli resource had a param for profile directory or profile target so that I can set up a different resource for dev and prod environments.

Also is there a way to access the dbt exit code?

I had a situation where the dagster run status was Success even though the dbt run failed because of bad credentials, so I've been using the run artifacts to confirm the run was successful:

def dbt_post_run_analysis(dbt_cli_task: DbtCliTask):
    try:
        run_results = dbt_cli_task.get_artifact("run_results.json")
        sys.stdout.write(pprint.pformat(run_results))
    except FileNotFoundError:
        raise Exception("DBT run results not found")
    
    for asset_result in run_results['results']:
        if asset_result['status'] != 'success':
            raise Exception("DBT run failed")

[jordandakota]

dbt_production_cli() looks great now. Very specific on guidance.

[geoHeil]

I am using now the following code

target = os.environ.get("DBT_TARGET", "local")
context.log.debug(f"Running dbt with target {target}")
dbt_cli_task = dbt.cli(
    ["run", "--target", target, "--profiles-dir", DBT_PROFILES_DIR],
    manifest=manifest,
    context=context,
)

@rexledesma : As discussed, I do not like this approach. I prefer the resources (which are validated for correctness during the startup of the job) and would prefer if the new DBT API would also be callable via the resources way natively. You mentioned this might be part of this weeks release - is this still true?

[rexledesma]

I am trying to avoid the scenario where the DbtCli is duplicating the configuration from dbt. This problem was prevalent in the previous iteration of the dbt resource.

• There is a conflict between Dagster's native way of configuration and dbt's native way of configuration. If the configuration exists in multiple places, new users of dbt and Dagster will need clarification, which is not a good experience.
• Ideally, configuring --target, --profile, and --profile-dir should be configurable in the existing global_config: I've filed [CT-2649] [Feature] Move --target and --profile command line flags to global config dbt-labs/dbt-core#7798 in dbt as a feature request.

[rexledesma]

Thanks for the feedback y'all!

I ended up changing my mind: profile and target will now be configuration on the DbtCli: #14627

[rexledesma]

Another follow-up on this: an exception will now be raised if the dbt CLI process fails with a non-zero exit code. But this behavior can be customized by using dbt.cli(..., raise_on_error=False). This is implemented in #15055.

This way, users can still customize how they want to exceptions to be handled in their dbt invocations.

[jordandakota]

I noticed you mentioned that as a dbt user I "want to declaratively schedule my assets (node_info_to_freshness_policy_fn, node_info_to_auto_materialize_policy_fn)". How does this play into the new @dbt_assets decorator? Do we define auto materialize there? I'm hoping this makes it easier to declaratively schedule my entire dbt graph, and preferably, also be able to access the dbt sources and set up source observations.

[rexledesma]

Hey Jordan,

We expect users to define their declarative scheduling configuration in the metadata of their dbt model definitions, which are written in .sql. From that perspective, node_info_to_freshness_policy_fn and node_info_to_auto_materialize_policy_fn will not be required, since we're just reading from your dbt project configuration.

So we expect a configuration like this on your model. This functionality to read from configuration exists today (you can check out the existing docstring: this is what the current functions are doing).

{{ config(
    dagster_freshness_policy={
        "maximum_lag_minutes": 123,
        "cron_schedule": "0 9 * * *",
        "cron_schedule_timezone": "America/New_York",
    },
    dagster_auto_materialize_policy={
        "type": "lazy"
    }
  )
}}

We're planning to further standardize this in dbt's meta field. This way, you can take advantage of dbt's native config inheritance when setting these values for groups of models.

[rexledesma]

This is now implemented in #14776.

[dveloz]

Hi! I'm trying to adapt our project to use the new dbt API but I'm having some issues when trying to materialize the events generated by DbtCli.
With the old API we were able to log materializations with code similar to the following

@op(
    required_resource_keys={"dbt"},
)
def custom_dbt_build_op(context):
    # run the command
    output = context.resources.dbt.cli(command="build")
    manifest = context.resources.dbt.get_manifest_json()

    # logs all materializations to Dagster
    events = generate_events(
        dbt_output=output,
        manifest_json=manifest,
    )
    for event in events:
        context.log_event(event)

    return output

My question is how to log the materializations events using the new DbtCli object. I have tried using the to_default_asset_events function to generate the events but I get the following error ,dagster._check.CheckError: Failure condition: Unexpected event <dagster._core.definitions.events.Output object at 0x00000193E2AC9BA0> ,
when trying to log them using context.log(event)

[geoHeil]

https://georgheiler.com/2023/06/13/unlocking-advanced-metadata-extraction-with-the-new-dbt-api-in-dagster/

[rexledesma]

Hi @dveloz! I've added an example to the repository for your use case: #14790.

What you need to do is to map these Output's to AssetMaterialization's. This is implicitly done for you when you use @dbt_assets, but since you are still using ops and jobs, you will need to do this manually.

from dagster import AssetMaterialization, Output, job, op
from dagster_dbt.cli import DbtCli, DbtManifest

from ..constants import MANIFEST_PATH

manifest = DbtManifest.read(path=MANIFEST_PATH)


@op
def my_dbt_build_op(dbt: DbtCli):
    for dagster_event in dbt.cli(["build"], manifest=manifest).stream():
        if isinstance(dagster_event, Output):
            yield AssetMaterialization(
                asset_key=manifest.get_asset_key_by_output_name(dagster_event.output_name),
                metadata=dagster_event.metadata,
            )
        else:
            yield dagster_event

    yield Output(None)


@job
def my_dbt_job():
    my_dbt_build_op()

[andreqaugusto]

Hey! I started using the new API and noticed a couple of things missing (or I did not understood fully):

1. it seems I can't customize the asset groups function (node_info_to_group_fn argument from load_assets_from_dbt_manifest).
2. I can't capture the dbt logs to show in dagster as easy as it was with the capture_logs argument from DbtCliResource

[rexledesma]

1. it seems I can't customize the asset groups function (node_info_to_group_fn argument from load_assets_from_dbt_manifest).

Rather than programmatically defining groups using node_info_to_group_fn, we are looking to add the ability to customize groups for your dbt models using your dbt project configuration.

Currently, we're looking forward to a hierarchy with precedence like the following. Note, these changes have not landed yet.

1. If dagster.group is specified for the model's meta configuration, use that. This is implemented in feat(dbt): parse asset group from +meta #14863.

models:
  - name: users
    meta:
      dagster:
        group: my_group

2. Otherwise, if an dbt group is configured for a model, use that instead. This is implemented in [1.4] feat(dbt)!: parse asset group from +group #14802.

The existing default behavior is to use the model's subdirectory as the group name, but we are looking to deprecate this implicit behavior in favor of the user configuring either (1) or (2).

2. I can't capture the dbt logs to show in dagster as easy as it was with the capture_logs argument from DbtCliResource

dbt logs are now captured in the Dagster compute logs, rather than the Dagster event stream. This occurs by default -- if you toggle to the compute log view in your Dagster run, the logs should be visible.

[mjkanji]

@rexledesma Have these changes for setting the DBT group landed in a release yet? If not, what's the current group definition logic? (It seems to base the groups on the folder hierarchy -- i.e., the group name is the name of the folder the .sql file is in?) Is there any way currently to override this?

[rexledesma]

@mjkanji

1. The ability to set the Dagster group using dagster.group on a dbt model's meta configuration landed in dagster-dbt=0.19.12.
2. The ability to set the Dagster group using use native dbt groups has landed, but won't be released until dagster-dbt=0.20.0, since this is a breaking change.

So you can use (1) to accomplish your use case.

[mjkanji]

Awesome, thanks!

[slopp]

An end-to-end example of using the new decorator is now available in our demo project: https://github.com/dagster-io/hooli-data-eng-pipelines

Take a look specifically at: https://github.com/dagster-io/hooli-data-eng-pipelines/blob/master/hooli_data_eng/assets/dbt_assets.py

A few notes:

1. This example customizes the asset's metadata and description by implementing methods for DbtManifest
2. This example shows how to use dagster partition ranges with dbt vars
3. This example includes customization of the asset metadata materialization events, capturing and surfacing dbt errors, and streams dbt logs to the dagster event logs
4. The dbt project in the same demo repository includes auto-materialization and freshness policies for dbt
5. The project uses a duckdb warehouse for easy reproducibility, but also includes an example of snowflake (and how to switch between the two using dagster resources and dbt profiles)

[gmontanola]

This is truly awesome, @slopp ❤️ I'm learning a lot from this repo!

One thing, this is happening when I try to materialize a partitioned dbt asset (selecting a single partition or a time window):

I found it weird because you defined this here

🤔

[slopp]

🤦‍♂️I forgot that this was still a problem for using the project locally.

We have an open issue for this: #14916

I fixed it for snowflake manually but did not for duckdb. I can push an update so the duckdb io manager is patched which would allow the project to run locally.

Here is the Snowflake patch:

https://github.com/dagster-io/hooli-data-eng-pipelines/blob/36db582a10caf62ddc22aef877d51148e945bc23/hooli_data_eng/resources/warehouse.py#L28

[moleary-gsa]

Thanks for this example!

I am wondering if it's expected behaviour that definition metadata added in the custom CustomizedDbtManifest node_info_to_metadata method doesn't show up when materializing a downstream python asset that depends on a dbt asset?

When I click on the downstream python asset and materialise just it, the only metadata that I can access from the context.upstream (within the load input method of my io manager) is the dagster_dbt/manifest entry.

When I materialise the entire job the metadata I add in node_info_to_metadata does show up

FWIW I am using node_info_to_metadata as a direct replacement for node_info_to_definition_metadata_fn so maybe there is a different way of achieving this

[moleary-gsa]

My bad - that's exactly what the issue you linked is describing.

[timvdp]

Hello, here is an other feature request for the new API.

We are using state selector, specifically the source_status:fresher+.

The issue that we encounter with the new API is the ability to execute multiple dbt commands in the same target folder or to be able to set the target folder. In order to use source_status:fresher+ selector, dbt run is expecting sources.json file to be present in target folder in addition to pass an other sources.json file with the --state parameter. This is achieve by executing dbt source freshness before dbt run --select source_status:fresher+ --state path/to/previous/execution. We had to override multiple functions in order to achieve this.

See the following dbt documentation for reference :
freshness | dbt Developer Hub
Syntax overview | dbt Developer Hub
Best practices | dbt Developer Hub

Thanks

[rexledesma]

Thanks for the feedback @timvdp. Can you file a feature request so we can track the implementation of this desired behavior?

Some comments here:

• We're following guidance to isolate the execution of multiple dbt commands so that artifacts don't overwrite each other. To do this, we store the results of each dbt invocation in a dynamic target path.
• However, there are some cases (like yours, as well as partial parsing, which is integrated into the dagster-dbt framework in feat(dbt): propagate partial parsing to dynamic target #14762), where the dynamic target path needs to be prepopulated with other files to let commands execute properly.

I could potentially see DbtCli.cli take in a target_files argument that would let you prepopulate the dynamic target path with these files so that your use case works out of the box.

[timvdp]

@rexledesma we can already retrieve the files with the DbtCliTask.get_artifact. If we have a way to pass it to DbtCli.cli that would be great. Here is the feature request : #15293

[IsaacChanXi]

Hi all, I am wondering if future releases would include creating an upstream dependency between a DBT model and any arbitrary asset in Dagster.

For example, a dummy function like

@asset
def hello():
    print(2)

and having a DBT model have an upstream dependency on this hello asset.

If this is not supported, does anyone have some suggestions on a potential work around?

Thank you!

[rexledesma]

This is available already.

• You can currently model upstream, external dependencies to your dbt models using dbt sources
• Then, use Dagster to model the computations that materialize the external dependencies: [RFC] Improved APIs for defining dbt asset dependencies #14964

[IsaacChanXi]

Hi Rex, great, thanks! I came across an example in the dagster-dbt tutorial but didn't know that this feature can be extended to any arbitrary python function.

[MarekVigas]

It would be really nice to be able to pass dbt tags to dagster tags. My main problem is that I need to limit the number of concurrent dbt runs to 1. But I only want to limit dbt assets and leave the python dagster assets unlimited. I found about the limits with tags but I'm not able to set the tags. I'm using the latest version 1.4.2 and importing assets with @dbt_assets syntax.

[rexledesma]

You can set the tags when creating a schedule for your assets. For example, build_schedule_from_dbt_selection has a tags argument that you can populate to set tags on the underlying job. Then, you can apply a concurrency limit to that tag.

[MarekVigas]

I'm using the AutoMaterializePolicy eager on all dbt assets in combination with FreshnessPolicy and would love to avoid any cron schedules. Thanks for the option.

[shionw]

Tried using this out and I'm getting an import error, is there anything that I missed?

dagster._core.errors.DagsterImportError: Encountered ImportError: cannot import name 'DbtCli' from 'dagster_dbt.cli' (C:\Users\username\AppData\Local\Programs\Python\Python311\Lib\site-packages\dagster_dbt\cli\__init__.py)

I am using poetry, and these are the dependencies.

• python = "^3.11"
• dagster = "^1.4.5"
• dagster-webserver = "^1.4.5"
• dagster-dbt = "^0.20.5"
• dbt-core = "^1.6.0"

[rexledesma]

This RFC is out of date. The new API's have already been released in dagster>1.4.0 and dagster-dbt>0.20.0.

You can check the following resources for information on the new API's:

• dbt + Dagster tutorial
• dbt + Dagster reference
• dagster-dbt API docs