https://dagster.io/ logo
Join Slack
Powered by
Hi there :wave: We'd like to define two dbt `Sequ...
# integration-dbt
n
Hi there 👋 We'd like to define two dbt `Sequence[AssetsDefinition]`s using the v1.4 
@dbt_asset
decorator, let's call them 
dbt_assets_1
and 
dbt_assets_2
... We'd like 
dbt_assets_1
to use the dbt build command, and 
dbt_assets_2
to use the dbt run command... The problem is that 
dbt_assets_1
selects all dbt assets (
select="*"
), and 
dbt_assets_2
selects everything upstream of one model (
select="+my_one_model"
)... From this, I get the following error:
Copy code
dagster._core.errors.DagsterInvalidDefinitionError: Asset key AssetKey(['staging', 'base_foo__bar_baz_qux']) is defined multiple times. Definitions found in modules: dagster_integrations.assets.
What's the best (or at least a good) way around this? Thanks :)
r
Two options: • 
dbt_assets_1
should exclude 
+my_one_model
• Consolidate your asset definitions into one asset definition, and parametrize the computation. e.g. parametrize what dbt command should run, either 
build
or `run`: https://dagster.slack.com/archives/C04CW71AGBW/p1692803335814609?thread_ts=1692726761.281029&cid=C04CW71AGBW
n
Thanks @rex! #1 isn't an option because 
dbt_assets_1
needs to include everything in 
dbt_assets_2
I'll check out option #2... ty spinny
Not sure how to implement #2 here... I've got jobs.py that looks like this:
Copy code
from dagster import AssetSelection, Config, RunConfig, define_asset_job


class MyDbtConfig1(Config):
    use_test: bool


my_dbt_config_1 = RunConfig(ops={"jaffle_shop_dbt_assets": MyDbtConfig1(use_test=True)})


asset_selection_dbt_1 = AssetSelection.keys(["orders"]).upstream()
job_dbt_1 = define_asset_job(
    name="dbt_job_1", selection=asset_selection_dbt_1, config=my_dbt_config_1
)

asset_selection_dbt_2 = AssetSelection.keys(["customers"]).upstream()
job_dbt_2 = define_asset_job(
    name="dbt_job_2", selection=asset_selection_dbt_2
)

asset_selection_python = AssetSelection.groups("python")
job_python = define_asset_job(
    name="python_job", selection=asset_selection_python
)
But when I reload the definitions, I get this error:
Copy code
dagster._core.errors.DagsterInvalidConfigError: Invalid default_value for Field.
    Error 1: Received unexpected config entry "use_test" at path root:jaffle_shop_dbt_assets:config. Expected: "{ }".

  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_grpc/server.py", line 295, in __init__
    self._loaded_repositories: Optional[LoadedRepositories] = LoadedRepositories(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_grpc/server.py", line 139, in __init__
    loadable_targets = get_loadable_targets(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_grpc/utils.py", line 47, in get_loadable_targets
    else loadable_targets_from_python_module(module_name, working_directory)
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/workspace/autodiscovery.py", line 35, in loadable_targets_from_python_module
    module = load_python_module(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/code_pointer.py", line 135, in load_python_module
    return importlib.import_module(module_name)
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/importlib/__init__.py", line 127, in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
  File "<frozen importlib._bootstrap>", line 1030, in _gcd_import
  File "<frozen importlib._bootstrap>", line 1007, in _find_and_load
  File "<frozen importlib._bootstrap>", line 986, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 680, in _load_unlocked
  File "<frozen importlib._bootstrap_external>", line 850, in exec_module
  File "<frozen importlib._bootstrap>", line 228, in _call_with_frames_removed
  File "/Users/nicolasmay/Documents/Learning/Dagster/tutorial-dbt-dagster/jaffle_shop/jaffle_dagster/jaffle_dagster/definitions.py", line 11, in <module>
    defs = Definitions(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/definitions_class.py", line 432, in __init__
    self._created_pending_or_normal_repo = _create_repository_using_definitions_args(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/definitions_class.py", line 303, in _create_repository_using_definitions_args
    def created_repo():
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/decorators/repository_decorator.py", line 163, in __call__
    else CachingRepositoryData.from_list(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/repository_definition/repository_data.py", line 346, in from_list
    return build_caching_repository_data_from_list(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/repository_definition/repository_data_builder.py", line 245, in build_caching_repository_data_from_list
    resolved_job = unresolved_job_def.resolve(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/unresolved_asset_job_definition.py", line 217, in resolve
    return build_asset_selection_job(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/asset_layer.py", line 819, in build_asset_selection_job
    asset_job = build_assets_job(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/assets_job.py", line 257, in build_assets_job
    return graph.to_job(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/graph_definition.py", line 641, in to_job
    return JobDefinition.dagster_internal_init(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/job_definition.py", line 278, in dagster_internal_init
    return JobDefinition(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/decorator_utils.py", line 195, in wrapped_with_pre_call_fn
    return fn(*args, **kwargs)
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/job_definition.py", line 229, in __init__
    self._config_mapping = _config_mapping_with_default_value(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_core/definitions/job_definition.py", line 1120, in _config_mapping_with_default_value
    updated_fields[name] = Field(
  File "/Users/nicolasmay/.pyenv/versions/3.9.9/lib/python3.9/site-packages/dagster/_config/field.py", line 331, in __init__
    raise DagsterInvalidConfigError(
r
you need to add 
MyDbtConfig1
as config to your 
@dbt_assets
decorator. See the API examples: https://docs.dagster.io/_apidocs/libraries/dagster-dbt#dagster_dbt.dbt_assets
n
Aha! Thank youuuuu!
6 Views
Open in Slack
PreviousNext
Powered by