@rex Just saw this and I thought it's great for the dbt integration https://www.getdbt.com/blog/introducing-webhooks-in-dbt-cloud/

Hi , I am doing a migration from dagster version 0.15 to 1.1. An error that we face is with dbt model single and double quotes. Both dagster is using the same Postgres engine in ECS with DBT version 1.1 I launch ECS job as a part of Dagster pipeline and run DBT as a Task Definition. The data models (.sql files) with double quotes work fine with the old version of Dagster "0.15". However with the new version "1.1" it fails. Gives the error on the Column does not exist can any share hint of what could be the problem?

Anyone could share a functional project using

load_assets_from_dbt_manifest

with the

select

argument? We're trying to move out of

load_assets_from_dbt_project

because of loading times, but the transition is not easy. I'd like to see if there are working examples somewhere before I can tell if the problems are in my head or in the library.

How can I change the asset key which is created from dagster-dbt? I know there is the key prefix and this is applied nicely (to both sources and models of the dbt project). However, for the sources a middle key space is created as well (wrehouse (prefix) / middle key (only sources) / table name. How can I force such a middle key (schema) for the models as well?

it is possible to add a group_name direct in the config of an dbt model?

Hi! I want to include my DBT project to the Dagster deployed environment. I need to create 2 or more workspaces in dagster but it is not working. how should i configure it? Thanks! 🙂

hi was wondering if python models are natively supported in Dagster and if so, can it be done w/o the use of dataproc for bigquery?

Hey Folks, I've searched the documentation but can't for the life of me figure out why my sources description and column documentation isn't showing up in dagit. If my sources are defined in a

sources.yml

file in my dbt project and are named "my_sources" for example. Do I need to do anything extra when loading the project to get the columne descriptions (and source table description) to show up in dagit?

What is the reasoning for keying source dbt assets with source_name and table name rather than using the schema which is used in the keying of model assets? Suppose, an upstream asset that is pandas or airbyte based and uses schema + table, the lineage will be broken unless one happens to name the source_name the same as the schema. I know I can override this behavior in load_assets_from_dbt_project() but I don't want to follow my own standard that might break other integrations. Thoughts? https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-dbt/dagster_dbt/asset_utils.py#L32

I’m looking for suggestions and best practices for working with an external dbt repo and dagster and how to integrate them with GitHub actions: currently, I have a GitHub repo with only the existing dbt project and in a separate one the dagster logic. The dbt repo has a workflow which is triggered from a push event or a cronjob. The workflow is responsible for seeding, snapshotting, running and testing. If it is triggered by a push event, only the impacted models are run and tested, otherwise the whole graph is considered. Ideally, I would like to have the dbt workflow responsible only to validate the changed models and let dagster run the periodic cronjobs. I was curious to know how you’ve designed the communication and workflows between the 2 repos in a CI/CD mindset

Support Update Hey folks, Dagster maintainer here. Our dbt experts are currently out of office, so some of your questions might not be answered until next week. Thanks for your patience!

How to support key range here? for long term fast backfill (dagster supports it) partition_key_to_vars=lambda partition_key: {"incremental_partition_key": partition_key},

Recently added a second source schema to my

schema.yml

(so I have two sets of tables with identical names, in two different schemas) and now I'm getting errors of this type:

KeyError: 'schema1_schema2_tablename'
  File "/home/rasmus/.cache/pypoetry/virtualenvs/annotator-WMT6nzGq-py3.8/lib/python3.8/site-packages/dagster/_grpc/impl.py", line 130, in core_execute_run
    yield from execute_run_iterator(
  File "/home/rasmus/.cache/pypoetry/virtualenvs/annotator-WMT6nzGq-py3.8/lib/python3.8/site-packages/dagster/_core/execution/api.py", line 1096, in __iter__
    yield from self.execution_context_manager.prepare_context()
  File "/home/rasmus/.cache/pypoetry/virtualenvs/annotator-WMT6nzGq-py3.8/lib/python3.8/site-packages/dagster/_utils/__init__.py", line 536, in generate_setup_events
    obj = next(self.generator)
  File "/home/rasmus/.cache/pypoetry/virtualenvs/annotator-WMT6nzGq-py3.8/lib/python3.8/site-packages/dagster/_core/execution/context_creation_pipeline.py", line 324, in orchestration_context_event_generator
    context_creation_data = create_context_creation_data(
  File "/home/rasmus/.cache/pypoetry/virtualenvs/annotator-WMT6nzGq-py3.8/lib/python3.8/site-packages/dagster/_core/execution/context_creation_pipeline.py", line 142, in create_context_creation_data
    resource_keys_to_init=get_required_resource_keys_to_init(
  File "/home/rasmus/.cache/pypoetry/virtualenvs/annotator-WMT6nzGq-py3.8/lib/python3.8/site-packages/dagster/_core/execution/resources_init.py", line 381, in get_required_resource_keys_to_init
    get_required_resource_keys_for_step(pipeline_def, step, execution_plan)
  File "/home/rasmus/.cache/pypoetry/virtualenvs/annotator-WMT6nzGq-py3.8/lib/python3.8/site-packages/dagster/_core/execution/resources_init.py", line 415, in get_required_resource_keys_for_step
    input_def = solid_def.input_def_named(step_input.name)
  File "/home/rasmus/.cache/pypoetry/virtualenvs/annotator-WMT6nzGq-py3.8/lib/python3.8/site-packages/dagster/_core/definitions/node_definition.py", line 144, in input_def_named
    return self._input_dict[name]

something in the node naming scheme appears to blend the schemas inadvertently?

Guys, I’m trying to run a dbt pipeline in Dagster, where first I tried to define the assets by the load_assets_from_dbt_project function, but when I called the execution of the dbt asset, the Dagster didn’t recognize the dbt manifest. Then I tried to generate the asset by the load_assets_from_dbt_manifest function and now I’m getting this error in the attached image: This is my code:

import json
from dagster import (
    asset,
    Definitions, 
    load_assets_from_package_module
)

from dagster_dbt import load_assets_from_dbt_manifest
from dagster_airbyte_example.assets import airbyte
from .config.constants import DOMAIN, BI_OWNER

airbyte_assets = load_assets_from_package_module(
    airbyte,
    key_prefix="best_target"
)

manifest_file = "/home/raphael-pacheco/workspace/dagster-airbyte-example/best_target/target/manifest.json"

with open(manifest_file, encoding='utf-8') as f:
    manifest = json.load(f)

dbt_assets = load_assets_from_dbt_manifest(
    manifest,
    key_prefix="best_target",
)

defs = Definitions(
    assets=[*airbyte_assets] + dbt_assets,
)

What am I doing wrong? Thanks in advance

Hey! Im trying to setup dbt using dagster in a mono repo using poetry. Is there a example that specifies dependencies using the pyproject.toml format?

Hey guys, does

load_assets_from_dbt_manifest

already separate my dbt models with custom schemas as separate assets?

Hey team! is there a way to

define

an asset as a subset of a dbt loaded from manifest asset?

If I create a job from DBT assets and a single non DBT asset which should be run between two DBT assets will dagster handle this properly?

Hey team, I want to use the

load_from_manifest

function to load my dbt assets. However, I don’t commit dbt’s target file to my github repo. How can I use s3 to load and update my manifest.json file through dagster’s dbt runs?

Hello, I have described dbt_project.yml as follows and used a custom schema. • dbt_project.yml

models:
  modelA:  # project name
    modelA:
      .....
    modelB:
      +schema: B
      .....

Asset is automatically loaded in Data Lineage as follows. • data asset

model/modelA/model_a_table
model/modelB/model_b_table

I am having trouble linking the modelB side and the materialization of the Data Lineage, although I am doing dbt runs daily and materializing at the same time as follows.

dbt_output = context.resources.dbt.run()
        for materialization in generate_materializations(dbt_output):
            yield materialization

This is because the “node” in the dbt_output output results in model.modelA.model_a_table, model.project.model_b_table and therefore will be materialized as model.modelA.model_b_table. Is there a good way to tie the materialization of the results from the daily dbt run to the data lineage?

is there a way for dbt to have knowledge of a daily partitioned source asset in Dagster? I would like to build an incremental model that only does stale partitions, so it doesn't need to reprocess the entire asset when a new day is processed in the source asset

Hi! I'm uploading the manifest from a GCP bucket. It reads it but I get this error:

TypeError: '>' not supported between instances of 'str' and 'int'

Hello, I have a project in which I transfer data from Google Cloud Storage to Bigquery and then I need to transform this data into DBT. If I create an asset for the flow from GCS to BQ and fetch the DBT assets through the

load_assets_from_dbt_project

function, my lineage does not create a dependency between the two flows, much less define an execution order. So I tried creating op and jobs to force the dependencies. I was able to set the dependencies as expected, but I ended up losing DBT lineage visibility. How can I keep both scenarios? Below I will send the codes and images. Thank you very much! Code without defined dependencies:

# dagster-dbt-example/__init__.py

from dagster import (
    asset, 
    job,
    Definitions
)
from dagster_dbt import load_assets_from_dbt_project, dbt_cli_resource

from .config.constants import DBT_PROJECT_DIR, DBT_PROFILE_DIR
from .assets.gcs import Extract
from .config.schemas import resources

@asset(group_name="bronze", compute_kind="python")
def bigquery_table_from_gcs_uri():
    """Function to send data of GCS files to BigQuery"""
    return Extract(resource=resource).gcs_to_bq_from_uri()

dbt_assets = load_assets_from_dbt_project(
    project_dir=DBT_PROJECT_DIR, profiles_dir=DBT_PROFILE_DIR, key_prefix=["best_target"]
)

defs = Definitions(
    assets=[bigquery_table_from_gcs_uri] + dbt_assets,
    resources={
        "dbt": dbt_cli_resource.configured(
            {
                "project_dir": DBT_PROJECT_DIR,
                "profiles_dir": DBT_PROFILE_DIR
            }
        )
    },
)

And this is the example with op and jobs, defining the dependencies:

# dagster-dbt-example/dbt/__init__.py

from dagster_dbt import dbt_run_op
from dagster import asset, job

from ...ops import get_resource
from ...assets.gcs import Extract
from ...config.schemas import resources

@asset(group_name="bronze", compute_kind="python")
def bigquery_table_from_gcs_uri():
    """Function to send data of GCS files to BigQuery from gs uri"""
    for resource in resources:
        print("RESOURCE ->", resource)
        Extract(resource=resources[resource]).gcs_to_bq_from_uri()

@job(name="transformations", resource_defs=get_resource())
def dbt_pipeline():
    dbt_run_op()

@asset(
    metadata={"owner": "<mailto:bi.dev@neoway.com.br|bi.dev@neoway.com.br>", "domain": "neoway", "priority": 1},
    group_name="silver",
    compute_kind="dbt"
)
def run_dbt_instance(bigquery_table_from_gcs_uri):
    """Function that defines asset dependencies"""
    dbt_pipeline.execute_in_process()

# dagster-dbt-example/__init__.py

from dagster import (
    repository, 
    define_asset_job, 
    ScheduleDefinition,
    load_assets_from_package_module, 
    with_resources,
    asset, 
    job
)
from dagster_dbt import dbt_run_op

from .ops import get_resource
from .assets import dbt
from .config.constants import DBT_PROJECT_DIR, DBT_PROFILE_DIR

from .assets.gcs import Extract

from .config.schemas import resources

resource = resources["pessoa_indicadores_veiculares"]

@asset(group_name="bronze", compute_kind="python")
def bigquery_table_from_gcs_uri():
    """Function to send data of GCS files to BigQuery"""
    return Extract(resource=resource).gcs_to_bq_from_uri()

@job(name="transformations", resource_defs=get_resource())
def dbt_pipeline():
    dbt_run_op()

@repository
def repository():
    return [
        with_resources(
            definitions=load_assets_from_package_module(dbt),
            resource_defs=get_resource()
        ),
        ScheduleDefinition(
            job=define_asset_job("all_assets", selection=["*"]),
            cron_schedule="@daily"
        )
    ]

Again, thanks to anyone who can help me ;)

Hi, im having the same issue described here https://github.com/dagster-io/dagster/issues/10918 Are there any updates on the topic? Im using using

load_assets_from_dbt_manifest

to load everything since i have my dbt project in another repository.

Are there any good examples floating about RE: creating an observable source asset for a DBT source that generates a

DataVersion

based on a query to the table it's generated from? For example, getting the number of rows.

Anyone else get this when upgrading to dbt ~1.5?

dagster._core.errors.DagsterImportError: Encountered ImportError: `No module named 'dbt.main'` while importing module foo.bar.

Announcement: the current version of

dagster-dbt

is incompatible with the newly released dbt 1.5. We are shipping a fix to resolve this. In the meantime, if you are encountering an error with the integration, we suggest that you pin

dbt-core<1.5

!

is there a way to materialize multiple partitions in one run with dbt assets generated via

load_assets_from_dbt_manifest

?

Hello Everyone. I'm BK i currently work as a data engineer for Paystack in Nigeria. My team currently orchestrates our daily jobs on dbt using dagster and i am looking to get up to speed with the system as i wasn't part of the team that set it up. I'll needing as much help as i can get as i try to add new features to improve on the existing system. Any pointers on the intersection between dagster and dbt is really appreciated.

Hello, I have two assets in my Dagster project. One is a Python routine that takes data from the GCS to BigQuery, the other is a DBT project that uses the tables that this Python routine generates. I am not able to make the DBT asset run only after the python asset executes. How can I define this dependency? Thank you for your attention!