在AWS Redshift的dbt中实现SCD2：如何定义条件自然键？

核心思路 #

先过滤出curr_recrd_flg = 'Y' and actve_flg = 'Y'的有效记录，再基于这些数据构建SCD2模型，避免无效数据干扰维度表的历史快照逻辑。

方法一：预处理源数据（推荐） #

在staging层提前过滤有效记录，再将其作为SCD2模型的输入，逻辑更清晰易维护：

1. 编写staging模型

-- models/staging/stg_crm_data.sql
select
    crm_id,
    -- 替换为你的实际业务字段
    customer_name,
    contact_email,
    curr_recrd_flg,
    actve_flg,
    current_timestamp as load_dt
from {{ source('your_source_schema', 'raw_crm_table') }}
-- 直接过滤出符合条件的有效记录
where curr_recrd_flg = 'Y' and actve_flg = 'Y'

2. 构建SCD2维度模型

方式A：增量模型+Merge策略

-- models/marts/dim_crm_scd2.sql
{{
    config(
        materialized='incremental',
        unique_key='crm_id', -- 过滤后同一crm_id仅一条有效记录，可简化唯一键
        incremental_strategy='merge',
        merge_update_columns=['customer_name', 'contact_email'] -- 指定需要更新的字段
    )
}}

with valid_source as (
    select * from {{ ref('stg_crm_data') }}
),

scd2_merge_result as (
    {{ dbt_utils.snapshot_merge_sql(
        target_schema=target.schema,
        target_table=this.name,
        source_name='valid_source',
        unique_key='crm_id',
        update_columns=['customer_name', 'contact_email'],
        invalidate_hard_deletes=True
    ) }}
)

select * from scd2_merge_result

方式B：dbt原生快照（标准SCD2实现）

创建快照文件：

# snapshots/crm_customer_snapshot.sql
{% snapshot crm_customer_scd2 %}

{{
    config(
        target_schema='snapshots',
        strategy='timestamp',
        unique_key='crm_id',
        updated_at='load_dt', -- 用staging层的加载时间作为更新判断依据
        invalidate_hard_deletes=True
    )
}}

-- 直接引用预处理后的有效数据
select * from {{ ref('stg_crm_data') }}

{% endsnapshot %}

方法二：在SCD2逻辑中直接过滤（不推荐） #

如果不想单独做staging层，也可以在增量模型的Merge语句中直接加入过滤条件：

-- models/marts/dim_crm_scd2.sql
{{
    config(
        materialized='incremental',
        unique_key='crm_id'
    )
}}

{% if is_incremental() %}

merge into {{ this }} as target
using (
    select * from {{ source('your_source_schema', 'raw_crm_table') }}
    where curr_recrd_flg = 'Y' and actve_flg = 'Y'
) as source
on target.crm_id = source.crm_id
-- 匹配到当前有效记录且字段有变化时，关闭旧版本
when matched and target.dbt_valid_to is null and (
    target.customer_name != source.customer_name or 
    target.contact_email != source.contact_email
) then
    update set dbt_valid_to = current_timestamp
-- 未匹配到则插入新版本
when not matched then
    insert (
        crm_id, customer_name, contact_email, 
        dbt_valid_from, dbt_valid_to
    )
    values (
        source.crm_id, source.customer_name, source.contact_email,
        current_timestamp, null
    )

{% else %}

-- 全量初始化时过滤有效记录
select
    crm_id,
    customer_name,
    contact_email,
    current_timestamp as dbt_valid_from,
    null as dbt_valid_to
from {{ source('your_source_schema', 'raw_crm_table') }}
where curr_recrd_flg = 'Y' and actve_flg = 'Y'

{% endif %}

注意事项 #

• 若源数据中同一crm_id存在多条curr_recrd_flg = 'Y' and actve_flg = 'Y'的记录，需先去重，否则会导致SCD2模型出现重复的当前版本记录。
• AWS Redshift的Merge操作对大表性能友好，但需确保unique_key字段有合适的索引。

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