Setting the file. One moment.
Subchapter 2.118
references/integrations/dagster-dbt/component-based-integration.mdMarkdown6 KBView on GitHub
The Component-based approach uses DbtProjectComponent to define dbt assets via YAML configuration.
This is the recommended approach for new projects.
For initial setup and scaffolding, see Scaffolding.
DbtProjectComponent is a StateBackedComponent that automatically compiles and caches your dbt
project’s manifest. See
StateBackedComponents for general
state management patterns.
What state is managed: The dbt manifest.json file, which contains all dbt models, tests,
sources, and their relationships.
How it’s compiled: Runs dbt parse (and dbt deps if needed) to generate the manifest.
Configuration: Use prepare_if_dev setting to control whether manifest is recompiled during
local development (defaults to true).
In CI/CD, use dg utils refresh-defs-state or dg plus deploy refresh-defs-state to compile the
manifest before deployment.
Use dbt selection syntax to filter which models are included:
attributes:
select: "tag:daily"
exclude: "tag:deprecated"Customize the dbt command executed for asset materialization:
attributes:
cli_args:
- build
- --full-refreshCLI args support template variables for dynamic values based on execution context.
For basic metadata customization, use the translation block:
attributes:
translation:
group_name: analytics
description: "dbt model {{ node.name }}"Template variables have access to the dbt node properties.
For complex customization, create a subclass and override get_asset_spec():
from collections.abc import Mapping
from typing import Any, Optional
import dagster as dg
from dagster_dbt import DbtProject, DbtProjectComponent
class CustomDbtComponent(DbtProjectComponent):
def get_asset_spec(
self,
manifest: Mapping[str, Any],
unique_id: str,
project: Optional[DbtProject],
) -> dg.AssetSpec:
base_spec = super().get_asset_spec(manifest, unique_id, project)
dbt_props = self.get_resource_props(manifest, unique_id)
return base_spec.merge_attributes(metadata={"model_name": dbt_props["name"]})Then use your custom component in defs.yaml:
type: my_project.components.custom_dbt_component.CustomDbtComponentDefine custom metadata in your dbt project files that can be consumed by your custom
get_asset_spec() implementation:
models:
- name: customers
meta:
custom_field: custom_valueThis metadata is accessible via the manifest in your translator code.
To partition incremental dbt models:
post_processingcli_argstemplate_vars_module: .template_vars
attributes:
cli_args:
- build
- --vars:
min_date: "{{ context.partition_time_window.start.strftime('%Y-%m-%d') }}"
max_date: "{{ context.partition_time_window.end.strftime('%Y-%m-%d') }}"
post_processing:
assets:
- target: "*"
attributes:
partitions_def: "{{ my_partitions_def }}"The context.partition_time_window variable is available in cli_args during execution. Dagster
automatically converts the vars dict to JSON format for dbt CLI.
For multiple partitions definitions, create separate DbtProjectComponent instances and use
select to filter models for each.
Enable automatic metadata fetching during materialization:
attributes:
include_metadata:
- row_count
- column_metadatarow_count: Fetch row counts for tablescolumn_metadata: Fetch column schema and lineage (extracted via sqlglot parsing)Both metadata types are fetched in parallel during dbt execution.
See Asset Checks for details on how dbt tests are loaded as Dagster asset checks.
See Dependencies for details on how Dagster parses dbt project dependencies and patterns for defining additional dependencies.
Use the asset_key_for_model method to reference dbt models from other components:
type: dagster.PythonScriptComponent
attributes:
script_path: export_customers.py
dependencies:
- "{{ context.load_component('dbt_component').asset_key_for_model('customers') }}"Use standard Dagster scheduling approaches with asset selections:
import dagster as dg
daily_dbt_job = dg.define_asset_job(
name="daily_dbt_models",
selection=dg.AssetSelection.groups("analytics"),
)
daily_schedule = dg.ScheduleDefinition(
job=daily_dbt_job,
cron_schedule="0 0 * * *",
)For declarative automation, configure AutomationConditions via custom get_asset_spec()
implementation.
Override the execute() method to customize execution behavior:
from collections.abc import Iterator
import dagster as dg
from dagster_dbt import DbtCliResource, DbtProjectComponent
class ConfigurableDbtComponent(DbtProjectComponent):
@property
def op_config_schema(self) -> type[dg.Config]:
class DbtConfig(dg.Config):
full_refresh: bool = False
return DbtConfig
def execute(self, context: dg.AssetExecutionContext, dbt: DbtCliResource) -> Iterator:
if context.op_config.get("full_refresh"):
args = ["build", "--full-refresh"]
else:
args = self.get_cli_args(context)
yield from dbt.cli(args, context=context).stream()Define op_config_schema property to specify available config options.
For dbt Cloud projects, see dbt Cloud Integration.