使用DBT快照构建SCD2维度及定制生成SQL的技术咨询

1. 重命名默认日期列（通过重写DBT宏实现） #

DBT快照生成的dbt_valid_from/dbt_valid_to列，可通过重写核心快照宏实现重命名，无需额外创建视图。核心要修改两个宏：snapshot_merge_sql（负责生成合并逻辑）和snapshot_staging_table（负责生成临时中间表）。

在项目的macros/目录下新建custom_scd2_snapshot.sql文件，写入以下代码：

-- 重写快照临时表宏，替换列名别名
{% macro snapshot_staging_table(strategy, source_sql, target_relation) %}
    with snapshot_query as (
        {{ source_sql }}
    ),
    ranked_snapshot as (
        select
            *,
            {{ strategy.unique_key }} as dbt_unique_key,
            {{ strategy.updated_at }} as dbt_updated_at,
            {{ strategy.updated_at }} as dbt_valid_from,
            cast(null as {{ dbt.type_timestamp() }}) as dbt_valid_to
        from snapshot_query
    )
    select
        *,
        dbt_valid_from as start_date,
        dbt_valid_to as end_date
    from ranked_snapshot
{% endmacro %}

-- 重写合并SQL宏，替换merge逻辑中的列名引用
{% macro snapshot_merge_sql(target, source, insert_cols) %}
    merge into {{ target }} as DBT_INTERNAL_DEST
    using {{ source }} as DBT_INTERNAL_SOURCE
    on DBT_INTERNAL_DEST.{{ target.unique_key }} = DBT_INTERNAL_SOURCE.dbt_unique_key

    when matched
        and DBT_INTERNAL_DEST.end_date is null
        and DBT_INTERNAL_SOURCE.dbt_updated_at > DBT_INTERNAL_DEST.dbt_updated_at
    then update
        set end_date = DBT_INTERNAL_SOURCE.start_date,
            current_indicator = 'N'

    when not matched
    then insert ({{ insert_cols | join(', ') }})
    values ({{ insert_cols | map('replace', 'dbt_valid_from', 'start_date') | map('replace', 'dbt_valid_to', 'end_date') | join(', ') }})
{% endmacro %}

2. 用自定义日期参数填充日期列 #

在快照配置中传入自定义日期变量，同时修改宏中的日期赋值逻辑，替换默认的系统时间函数。

步骤1：在快照文件中定义自定义日期变量 #

比如在snapshots/dim_customers.sql中：

{% snapshot dim_customers_scd2 %}
{{
    config(
        target_schema='snapshots',
        strategy='timestamp',
        unique_key='customer_id',
        updated_at='last_modified_date',
        vars={'snapshot_run_date': '2024-05-20 00:00:00'}
    )
}}

select * from {{ source('raw', 'customers') }}
{% endsnapshot %}

步骤2：修改宏中的日期赋值逻辑 #

在之前的custom_scd2_snapshot.sql里，把dbt_valid_from的赋值替换为自定义变量：

-- 修改ranked_snapshot中的dbt_valid_from赋值
{{ var('snapshot_run_date') }} as dbt_valid_from,

如果需要动态传入日期，可在执行命令时传入：

dbt snapshot --vars '{"snapshot_run_date": "2024-05-20 00:00:00"}'

3. 新增current_indicator列并标记最新记录 #

需要在临时表和合并逻辑中添加该列，确保最新记录标记为'Y'，历史过期记录标记为'N'。

修改宏添加current_indicator列 #

在custom_scd2_snapshot.sql的snapshot_staging_table宏中，新增列定义：

select
    *,
    dbt_valid_from as start_date,
    dbt_valid_to as end_date,
    'Y' as current_indicator
from ranked_snapshot

完善合并逻辑中的更新规则 #

在snapshot_merge_sql宏的update分支中，添加对current_indicator的更新：

when matched
    and DBT_INTERNAL_DEST.end_date is null
    and DBT_INTERNAL_SOURCE.dbt_updated_at > DBT_INTERNAL_DEST.dbt_updated_at
then update
    set end_date = DBT_INTERNAL_SOURCE.start_date,
        current_indicator = 'N'

初始化注意事项 #

首次运行快照前，若目标表已存在，需手动添加current_indicator列；若为全新快照，DBT会自动根据宏生成的列创建表。

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