DBT增量模型中左连接后标记并删除互补表关联行的实现方案

核心方案 #

通过封装Jinja宏，在单次关联操作中完成主表更新、互补表匹配标记和已匹配行删除，避免重复连接带来的性能开销。核心思路是先通过一次连接提取所有匹配的关联键存入临时表，后续操作均基于该临时表执行，无需重复关联两张主表。

1. 编写处理宏 #

以下宏适用于dbt（Data Build Tool）场景，可直接在模型中调用：

{% macro process_complementary_table(source_table1, source_table2, join_key) %}
    -- 1. 一次性提取所有匹配的关联键，存入临时表
    CREATE TEMP TABLE matched_keys AS
    SELECT DISTINCT tab2.{{ join_key }}
    FROM {{ source_table1 }} tab1
    INNER JOIN {{ source_table2 }} tab2
        ON tab1.{{ join_key }} = tab2.{{ join_key }};

    -- 2. 更新主表table1：用互补表数据覆盖匹配行，未匹配行保留原数据
    {% if execute %}
        -- 若存在匹配数据则执行更新，否则跳过
        {% set has_matches = run_query("SELECT COUNT(*) FROM matched_keys").columns[0].values()[0] %}
        {% if has_matches > 0 %}
            UPDATE {{ source_table1 }} tab1
            SET
                col_2 = tab2.col_2,
                col_3 = tab2.col_3
            FROM {{ source_table2 }} tab2
            WHERE tab1.{{ join_key }} = tab2.{{ join_key }};
        {% endif %}
    {% endif %}

    -- 3. 标记互补表中已匹配的行
    UPDATE {{ source_table2 }} tab2
    SET col_4 = 1
    WHERE EXISTS (
        SELECT 1 FROM matched_keys mk
        WHERE mk.{{ join_key }} = tab2.{{ join_key }}
    );

    -- 4. 删除互补表中已标记的行
    DELETE FROM {{ source_table2 }}
    WHERE col_4 = 1;

    -- 清理临时表
    DROP TABLE IF EXISTS matched_keys;
{% endmacro %}

2. 在主模型中调用宏 #

直接在dbt模型文件中调用上述宏，传入数据源和关联键：

{{ process_complementary_table(
    source_table1=source('xxx','table1'),
    source_table2=source('xxx','table2'),
    join_key='col_1'
) }}

3. 结果验证 #

执行后将得到预期结果：

• 更新后的table1：

col_1       col_2         col_3       col_4
10          'Goodbye'     'Earth'     'Hi'
20          'Hello'       'World'     'Hi'

• 最终table2：

col_1       col_2       col_3       col_4
30          'Goodbye'   'Earth'     null

方案优势 #

• 仅执行一次INNER JOIN提取匹配键，后续标记、删除操作均基于临时表，避免重复关联两张大表的性能损耗
• 宏封装后可复用在多个类似场景中，减少代码冗余
• 加入了匹配数据判断，无匹配时跳过主表更新，避免无效操作

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