如何在dbt模型中使用映射表动态修改列名

核心思路 #

利用dbt的Jinja模板能力，读取列名映射表的对应关系，动态生成旧列名 AS 新列名的SQL语句，实现批量替换列名。

步骤1：准备列名映射表 #

先在数据库中创建或维护一个列名映射表（也可以直接在dbt模型中定义静态映射），结构示例如下：

-- 示例映射表（可存储为dbt模型或直接在数据库中创建）
select 
  'old_user_id' as old_column_name,
  'user_id' as new_column_name
union all
select 
  'old_order_amount' as old_column_name,
  'order_total' as new_column_name
union all
select 
  'old_create_time' as old_column_name,
  'created_at' as new_column_name

步骤2：编写dbt动态映射模型 #

如果映射表是数据库中的实体表（或dbt模型），可以用以下代码动态生成字段映射：

-- 读取映射表数据
{% set mapping_query %}
select old_column_name, new_column_name from {{ ref('column_mapping_table') }}
{% endset %}

{% set mapping_results = run_query(mapping_query) %}

-- 仅在运行阶段执行映射逻辑，避免编译报错
{% if execute %}
  {% set mapping_dict = mapping_results.rows | map('dict') | list %}
{% else %}
  {% set mapping_dict = [] %}
{% endif %}

-- 动态生成字段映射SQL
select
  {% for item in mapping_dict %}
    -- 适配带特殊字符的列名，用adapter.quote处理不同数据库的引号规则
    {{ adapter.quote(item.old_column_name) }} as {{ adapter.quote(item.new_column_name) }}
    {% if not loop.last %},{% endif %}
  {% endfor %}
from {{ ref('original_old_column_table') }}

静态映射替代方案 #

如果列名映射不会频繁变动，也可以直接在模型中定义静态字典，省去查询映射表的步骤：

{% set mapping_dict = [
  {'old_column_name': 'old_user_id', 'new_column_name': 'user_id'},
  {'old_column_name': 'old_order_amount', 'new_column_name': 'order_total'},
  {'old_column_name': 'old_create_time', 'new_column_name': 'created_at'}
] %}

select
  {% for item in mapping_dict %}
    {{ adapter.quote(item.old_column_name) }} as {{ adapter.quote(item.new_column_name) }}
    {% if not loop.last %},{% endif %}
  {% endfor %}
from {{ ref('original_old_column_table') }}

注意事项 #

• 确保映射表覆盖所有需要修改的列；如果只需要替换部分列，可在循环后手动追加其他字段
• adapter.quote函数可以适配不同数据库（如Snowflake、BigQuery、PostgreSQL）的列名引号规则，避免特殊字符列名报错
• execute变量用于区分dbt的编译阶段和运行阶段，防止编译时因无法读取数据库数据而报错

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