如何将含Delete+Insert的Snowflake存储过程转换为DBT模块？

问题背景 #

我正在学习DBT，希望将以下Snowflake存储过程改写为DBT模型，但不清楚如何在DBT中实现SQL的Delete/Insert逻辑。

原Snowflake存储过程代码：

create or replace procedure staging.ingest_google_campaigns_into_master()
    returns varchar
    language sql
as
$$
    begin
    DELETE FROM GOOGLE_ADWORD_CAMPAIGN
           WHERE DT IN (SELECT DISTINCT ORIGINALDATE AS DT FROM GOOGLEADWORDS_CAMPAIGN);
    INSERT INTO GOOGLE_ADWORD_CAMPAIGN
        SELECT DISTINCT *
        FROM
            (
               SELECT  g.* ,
                       YEAR(TO_TIMESTAMP(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR)) AS YEAR,
                       LPAD(MONTH(TO_TIMESTAMP(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR)),2,0) AS MONTH,
                       LPAD(DAY(TO_TIMESTAMP(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR)),2,0) AS DAY,
                       TO_DATE(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR) AS DT

               FROM GOOGLEADWORDS_CAMPAIGN g
            ) t;
    end;
$$
;

该过程先删除GOOGLE_ADWORD_CAMPAIGN表中的旧数据（匹配GOOGLEADWORDS_CAMPAIGN里的ORIGINALDATE对应的DT），再插入新的数据集。

解决方案：用DBT增量模型实现先删后插逻辑 #

DBT中实现这类"删除重复日期数据再插入"的逻辑，最适合用增量模型（Incremental Model），配合自定义删除逻辑替代存储过程的Delete+Insert步骤。

步骤1：创建增量模型文件（示例路径：models/staging/google_adword_campaign.sql）

{{ config(
    materialized='incremental',
    unique_key='DT' -- 可根据业务实际调整为唯一键或组合键
) }}

WITH transformed_data AS (
    SELECT DISTINCT 
        g.*,
        YEAR(TO_TIMESTAMP(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR)) AS YEAR,
        LPAD(MONTH(TO_TIMESTAMP(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR)),2,0) AS MONTH,
        LPAD(DAY(TO_TIMESTAMP(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR)),2,0) AS DAY,
        TO_DATE(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR) AS DT
    FROM GOOGLEADWORDS_CAMPAIGN g
)

-- 增量运行时，先删除目标表中与待插入数据重复的DT记录
{% if is_incremental() %}
DELETE FROM {{ this }}
WHERE DT IN (SELECT DISTINCT DT FROM transformed_data);
{% endif %}

-- 插入转换后的数据
SELECT * FROM transformed_data

步骤2：关键配置说明

• materialized='incremental'：标记该模型为增量模型，首次运行会全量创建表，后续运行仅执行增量逻辑
• is_incremental()：DBT内置宏，判断当前是否为增量运行（首次运行跳过Delete语句）
• {{ this }}：指代当前模型对应的目标表（需在dbt_project.yml中配置schema与表名映射）

替代方案：用运行钩子（Run Hooks）执行删除逻辑

如果需要每次运行都先删后插全量数据，可在模型配置中添加pre-hook：

{{ config(
    materialized='table',
    pre_hook=[
        "DELETE FROM GOOGLE_ADWORD_CAMPAIGN WHERE DT IN (SELECT DISTINCT ORIGINALDATE AS DT FROM GOOGLEADWORDS_CAMPAIGN)"
    ]
) }}

WITH transformed_data AS (
    SELECT DISTINCT 
        g.*,
        YEAR(TO_TIMESTAMP(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR)) AS YEAR,
        LPAD(MONTH(TO_TIMESTAMP(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR)),2,0) AS MONTH,
        LPAD(DAY(TO_TIMESTAMP(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR)),2,0) AS DAY,
        TO_DATE(DATE_PART(EPOCH_SECOND, ORIGINALDATE::TIMESTAMP)::VARCHAR) AS DT
    FROM GOOGLEADWORDS_CAMPAIGN g
)

SELECT * FROM transformed_data

这种方式每次运行都会先执行Delete语句，再插入全量转换后的数据，适合无需增量更新、每次替换指定日期数据的场景。

注意事项

• 确保GOOGLEADWORDS_CAMPAIGN是DBT可访问的数据源（需在profiles.yml配置Snowflake连接，或通过sources.yml定义源表）
• 若DT不是唯一键，需调整unique_key为实际业务唯一键组合，避免数据重复

内容的提问来源于stack exchange，提问作者Paweł Adamski