如何在dbt中按文件夹定义变量及复用实体代码？

1. 在dbt_project.yaml中实现层级变量结构 #

通过嵌套键值对构建层级结构，将各实体的专属变量归到统一父键下，同时可保留全局共享变量，示例如下：

vars:
  # 全局共享变量
  shared_db: "analytics_platform"
  # 实体专属层级变量
  entities:
    bank:
      target_schema: "bank_transactions"
      source_table: "raw_bank_data"
      table_suffix: "_txn"
    names:
      target_schema: "customer_names"
      source_table: "raw_customer_names"
      table_suffix: "_profile"
    cars:
      target_schema: "vehicle_records"
      source_table: "raw_vehicle_data"
      table_suffix: "_info"

这种结构清晰区分共享配置与实体个性化配置，便于集中维护。

2. 在各类文件中引用层级变量 #

模型SQL文件中引用 #

直接通过点语法或键索引访问层级变量，例如在bank.sql模型中：

{{ config(schema=var('entities').bank.target_schema) }}

select
  id,
  transaction_date,
  amount
from {{ source(var('shared_db'), var('entities').bank.source_table) }}

若需批量生成实体模型，可结合宏遍历变量：

-- 宏示例：遍历所有实体生成模型
{% macro generate_all_entity_models() %}
  {% for entity_name, entity_config in var('entities').items() %}
    {{ config(schema=entity_config.target_schema) }}
    select * from {{ source(var('shared_db'), entity_config.source_table) }}
  {% endfor %}
{% endmacro %}

YAML配置文件中引用 #

在schema.yml或模型配置块中，同样用var()函数访问层级变量：

models:
  my_dbt_project:
    bank:
      +schema: "{{ var('entities').bank.target_schema }}"
      columns:
        - name: id
          tests:
            - unique

3. 其他处理多实体通用代码的思路 #

模型继承 #

创建通用基础模型，各实体模型通过继承复用逻辑，仅传递专属参数：

-- models/base_entity.sql
{{ config(
    schema=var('target_schema'),
    alias=var('entity_name') ~ var('table_suffix')
) }}
select
  id,
  created_at,
  updated_at
from {{ source(var('shared_db'), var('source_table')) }}

在bank.sql中继承并传参：

{{ config(
    target_schema=var('entities').bank.target_schema,
    entity_name='bank',
    table_suffix=var('entities').bank.table_suffix,
    source_table=var('entities').bank.source_table
) }}
{{ ref('base_entity') }}

宏封装通用逻辑 #

将实体的通用处理逻辑封装为宏，各实体模型直接调用宏并传入自身配置：

-- macros/entity_transform.sql
{% macro entity_transform(entity_config) %}
{{ config(schema=entity_config.target_schema) }}
select
  id,
  {{ dbt_utils.star(from=source(var('shared_db'), entity_config.source_table), except=['raw_id']) }}
from {{ source(var('shared_db'), entity_config.source_table) }}
{% endmacro %}

在cars.sql中调用：

{{ entity_transform(var('entities').cars) }}

种子文件管理实体配置 #

将实体配置存入种子文件（如data/entities_config.csv），通过宏读取种子数据动态生成模型，避免在yaml中维护大量嵌套配置：

entity_name,target_schema,source_table,table_suffix
bank,bank_transactions,raw_bank_data,_txn
names,customer_names,raw_customer_names,_profile
cars,vehicle_records,raw_vehicle_data,_info

编写宏遍历种子数据生成模型：

{% macro generate_models_from_seed() %}
  {% set entities = dbt_utils.get_column_values(table=ref('entities_config'), column='entity_name') %}
  {% for entity in entities %}
    {% set config = dbt_utils.get_row_values(table=ref('entities_config'), filter_column='entity_name', filter_value=entity) %}
    {{ config(schema=config.target_schema) }}
    select * from {{ source(var('shared_db'), config.source_table) }}
  {% endfor %}
{% endmacro %}

环境变量与变量覆盖 #

针对不同环境（开发/生产）的配置差异，可在profiles.yml或环境变量中覆盖特定实体变量，例如生产环境的profiles.yml：

my_prod_profile:
  outputs:
    prod:
      type: snowflake
      ...
      vars:
        entities:
          bank:
            target_schema: "prod_bank_transactions"

内容的提问来源于stack exchange，提问作者bellotto