https://dagster.io/ logo
#integration-dbt
Title
# integration-dbt
c

Constantin Lungu

03/21/2023, 11:13 PM
Hello everyone! I'm trying to get started using
dagster_dbt
. I'm encountering the following error:
Copy code
agster._core.errors.DagsterInvalidDefinitionError: resource with key 'dbt' required by op 'run_dbt_da47d' was not provided. Please provide a <class 'dagster._core.definitions.resource_definition.ResourceDefinition'> to key 'dbt', or change the required key to one of the following keys which points to an <class 'dagster._core.definitions.resource_definition.ResourceDefinition'>: ['io_manager']
I see that
Copy code
dbt_cli_resource.configured(
        {"project_dir": DBT_PROJECT_DIR, "profiles_dir": DBT_PROFILES_DIR}
    )
Is already a ResourceDefinition.
Copy code
import os

from dagster_dbt import dbt_cli_resource, load_assets_from_dbt_project
from dagster_duckdb_pandas import duckdb_pandas_io_manager

from dagster import (
    Definitions,
    ScheduleDefinition,
    define_asset_job,
    fs_io_manager,
    load_assets_from_package_module,
)
from dagster._utils import file_relative_path

DBT_PROJECT_DIR = file_relative_path(__file__, "../dbt_project")
DBT_PROFILES_DIR = file_relative_path(__file__, "../dbt_project/config")

# all assets live in the default dbt_schema
dbt_assets = load_assets_from_dbt_project(
    DBT_PROJECT_DIR,
    DBT_PROFILES_DIR,
    # prefix the output assets based on the database they live in plus the name of the schema
    key_prefix=["duckdb", "dbt_schema"],
    # prefix the source assets based on just the database
    # (dagster populates the source schema information automatically)
    source_key_prefix=["duckdb"],
)

a = dbt_cli_resource.configured(
        {"project_dir": DBT_PROJECT_DIR, "profiles_dir": DBT_PROFILES_DIR})

print(type(a))

resources = {
    # this io_manager allows us to load dbt models as pandas dataframes
    "io_manager": duckdb_pandas_io_manager.configured(
        {"database": os.path.join(DBT_PROJECT_DIR, "example.duckdb")}
    ),
    # this resource is used to execute dbt cli commands
    "dbt": dbt_cli_resource.configured(
        {"project_dir": DBT_PROJECT_DIR, "profiles_dir": DBT_PROFILES_DIR}
    )
}

everything_job = define_asset_job("everything_everywhere_job", selection="*")

defs = Definitions(
    assets=[*dbt_assets],
    resources=resources,
    schedules=[
        ScheduleDefinition(job=everything_job, cron_schedule="@weekly")    ]
)
Can someone please point out what am I missing?
j

jamie

03/22/2023, 2:23 PM
Hey @Constantin Lungu how are you materializing the dbt assets (through the ui, cli, etc)? also cc @rex @owen for visibility
r

rex

03/22/2023, 2:42 PM
Hi, I’m unable to replicate this. I edited the
tutorial_dbt_dagster
project with your sample code and I’m not getting your error.
Copy code
dagster project from-example --name tutorial_dbt_dagster --example tutorial_dbt_dagster
Copy code
import os
from dagster import (
    Definitions,
    load_assets_from_modules,
    define_asset_job,
    ScheduleDefinition,
)
from dagster_dbt import dbt_cli_resource
from dagster_duckdb_pandas import duckdb_pandas_io_manager

from tutorial_dbt_dagster import assets
from tutorial_dbt_dagster.assets import DBT_PROFILES, DBT_PROJECT_PATH

resources = {
    "dbt": dbt_cli_resource.configured(
        {
            "project_dir": DBT_PROJECT_PATH,
            "profiles_dir": DBT_PROFILES,
        },
    ),
    "io_manager": duckdb_pandas_io_manager.configured(
        {"database": os.path.join(DBT_PROJECT_PATH, "tutorial.duckdb")}
    ),
}

everything_job = define_asset_job("everything_everywhere_job", selection="*")

defs = Definitions(
    assets=load_assets_from_modules([assets]),
    resources=resources,
    schedules=[ScheduleDefinition(job=everything_job, cron_schedule="@weekly")],
)
Screenshot 2023-03-22 at 10.42.54 AM.png
could you confirm that you’ve reloaded your code definitions? your setup looks right to me.
j

jamie

03/22/2023, 2:44 PM
also what version of dagster are you using? there was a bug a while back that made use erroneously determine which resources were available in some very specific circumstances
c

Constantin Lungu

03/22/2023, 2:47 PM
This is the version of my packages. So
dagster-1.2.2
j

jamie

03/22/2023, 2:48 PM
ok - the bug would have been fixed by then. I think it was back in like 1.1.7 or something
D 1
c

Constantin Lungu

03/22/2023, 2:49 PM
@rex Yes I did reload code definitions, result still the same.
I did manage to get a working setup while following https://docs.dagster.io/integrations/dbt/using-dbt-with-dagster/part-one
So nevermind, I'm good now 🙂 Thanks for your help.
8 Views