使用DBT实现Redshift增量加载：聚合更新替代Upsert方案咨询

针对Redshift跨Schema的增量需求（存在userid则累加数值列，不存在则插入），可以通过自定义DBT增量模型的Merge逻辑实现，以下是具体步骤：

1. 基础增量模型配置 #

先定义模型的基础结构，指定增量模式、唯一键和目标Schema：

{{ config(
    materialized='incremental',
    unique_key='userid',
    target_schema='your_target_report_schema',
    incremental_strategy='merge'
) }}

-- 增量数据查询：从源Schema获取数据，增量场景下过滤新数据
SELECT
    userid,
    total_deposit,
    total_withdrawal,
    sync_time -- 假设源表有同步时间字段，用于增量过滤
FROM {{ source('your_source_schema', 'source_table') }}
{% if is_incremental() %}
    -- 仅加载上次同步后新增的数据
    WHERE sync_time > (SELECT COALESCE(MAX(sync_time), '1970-01-01') FROM {{ this }})
{% endif %}

2. 自定义Merge逻辑实现累加更新 #

DBT默认的Merge策略是替换字段值，我们需要覆盖默认逻辑，改为匹配时累加数值：

{{ config(
    materialized='incremental',
    unique_key='userid',
    target_schema='your_target_report_schema',
    incremental_strategy='merge',
    -- 自定义Merge SQL，实现累加逻辑
    merge_sql="""
        MERGE INTO {{ this }} AS target
        USING {{ temp_table }} AS source
        ON target.userid = source.userid
        WHEN MATCHED THEN
            UPDATE SET
                total_deposit = target.total_deposit + source.total_deposit,
                total_withdrawal = target.total_withdrawal + source.total_withdrawal,
                sync_time = source.sync_time -- 更新同步时间为最新值
        WHEN NOT MATCHED THEN
            INSERT (userid, total_deposit, total_withdrawal, sync_time)
            VALUES (source.userid, source.total_deposit, source.total_withdrawal, source.sync_time)
    """
) }}

-- 先聚合同userid的增量数据，避免多次累加
SELECT
    userid,
    SUM(total_deposit) AS total_deposit,
    SUM(total_withdrawal) AS total_withdrawal,
    MAX(sync_time) AS sync_time
FROM {{ source('your_source_schema', 'source_table') }}
{% if is_incremental() %}
    WHERE sync_time > (SELECT COALESCE(MAX(sync_time), '1970-01-01') FROM {{ this }})
{% endif %}
GROUP BY userid

3. 关键注意事项 #

• 增量数据聚合：如果源数据中同一个userid可能有多条增量记录，必须先通过GROUP BY userid聚合数值列，否则会导致同批次内重复累加。
• 首次同步处理：首次全量同步时目标表为空，用COALESCE给MAX(sync_time)设置默认值，确保过滤条件生效。
• 字段完整性：如果目标表有创建时间、更新时间等额外字段，需要在INSERT和UPDATE语句中对应补充。

4. 验证方法 #

执行dbt run后，通过以下SQL验证结果：

-- 检查已有用户的数值是否正确累加
SELECT userid, total_deposit, total_withdrawal 
FROM your_target_report_schema.t1 
WHERE userid = 'existing_user_id';

-- 检查新用户是否成功插入
SELECT userid, total_deposit, total_withdrawal 
FROM your_target_report_schema.t1 
WHERE userid = 'new_user_id';

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