如何在dbt中循环对多张表执行相似转换操作？

dbt本身没有原生的FOR LOOP语法，但可以通过**宏（Macros）**结合内置功能实现批量处理，以下是具体方案：

方法一：编写宏遍历指定表列表 #

通过自定义宏，用dbt的Jinja2循环语法遍历目标表，批量生成你需要的INSERT语句。

1. 定义宏 #

创建宏文件（比如macros/insert_missing_rows.sql），内容如下：

{% macro insert_missing_rows() %}
    -- 方式1：手动指定需要处理的表名
    {% set target_tables = ['user', 'order', 'product'] %}

    -- 方式2：动态获取loading schema下的所有表（自动遍历，无需手动维护列表）
    -- {% set loading_relations = dbt_utils.get_relations_by_pattern('loading', '%') %}
    -- {% set target_tables = loading_relations | map(attribute='name') | list %}

    {% for table_name in target_tables %}
        INSERT INTO stage.{{ table_name }}
        SELECT * FROM loading.{{ table_name }}
        EXCEPT
        SELECT * FROM stage.{{ table_name }};
    {% endfor %}
{% endmacro %}

2. 调用宏 #

在一个模型文件（比如models/stage/batch_insert.sql）中调用该宏：

{{ insert_missing_rows() }}

执行dbt run时，该模型会自动遍历所有目标表，执行对应的插入逻辑。

注意事项 #

• 确保loading和stage schema下的对应表结构完全一致，否则EXCEPT语句会因字段不匹配报错
• 针对大型表，EXCEPT的性能可能偏低，建议改用主键对比逻辑（比如WHERE NOT EXISTS）优化，示例：

  INSERT INTO stage.{{ table_name }}
  SELECT l.* 
  FROM loading.{{ table_name }} l
  WHERE NOT EXISTS (
      SELECT 1 FROM stage.{{ table_name }} s 
      WHERE s.id = l.id -- 替换为你的主键字段
  );
  

• 动态获取表名时，可通过添加过滤条件排除不需要处理的表，避免误操作

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