Oracle/MSSQL下dbt动态Pivot行转列实现求助

Oracle 解决方案 #

Oracle原生Pivot需显式指定列名，针对超100个动态列的场景，需结合dbt宏+动态SQL实现：

步骤1：创建获取动态列名的宏 #

在dbt项目macros目录下新建get_pivot_columns.sql：

{% macro get_pivot_columns() %}
    select listagg('''' || adco_field_name || ''' as ' || adco_field_name, ', ') within group (order by adco_field_name)
    from (
        select distinct af.adco_field_name
        from {{ source('adco', 'sow_src_data') }} sd
        join {{ source('adco', 'src_field') }} sf on sd.src_field_id = sf.src_field_id
        join {{ source('adco', 'src_mapping') }} sm on sm.src_field_id = sf.src_field_id
        join {{ source('adco', 'adco_field') }} af on af.adco_field_id = sm.adco_field_id
        where effective_date = '2024-02-08'
    )
{% endmacro %}

步骤2：主模型中使用动态Pivot #

{{ config(materialized='table') }}

declare
    v_pivot_cols varchar2(32767);
begin
    -- 获取动态列名
    select {{ get_pivot_columns() }} into v_pivot_cols from dual;

    -- 执行动态Pivot并生成目标表
    execute immediate '
        create table {{ this }} as
        with source_data as (
            select 
                sd.field_value,
                af.adco_field_name
            from 
                {{ source('adco', 'sow_src_data') }} sd
                join {{ source('adco', 'src_field') }} sf on sd.src_field_id = sf.src_field_id
                join {{ source('adco', 'src_mapping') }} sm on sm.src_field_id = sf.src_field_id
                join {{ source('adco', 'adco_field') }} af on af.adco_field_id = sm.adco_field_id
            where 
                effective_date = ''2024-02-08''
        )
        select *
        from source_data
        pivot (
            max(field_value) for adco_field_name in (' || v_pivot_cols || ')
        )
    ';
end;
/

MSSQL 解决方案 #

MSSQL可通过STRING_AGG生成动态列名，结合动态SQL实现：

主模型代码 #

{{ config(materialized='table') }}

declare @pivot_cols nvarchar(max);
declare @sql nvarchar(max);

-- 获取动态列名
select @pivot_cols = string_agg(quotename(adco_field_name), ', ')
from (
    select distinct af.adco_field_name
    from {{ source('adco', 'sow_src_data') }} sd
    join {{ source('adco', 'src_field') }} sf on sd.src_field_id = sf.src_field_id
    join {{ source('adco', 'src_mapping') }} sm on sm.src_field_id = sf.src_field_id
    join {{ source('adco', 'adco_field') }} af on af.adco_field_id = sm.adco_field_id
    where effective_date = '2024-02-08'
) as cols;

-- 拼接并执行动态SQL
set @sql = N'
    with source_data as (
        select 
            sd.field_value,
            af.adco_field_name
        from 
            ' + {{ source('adco', 'sow_src_data') }} + N' sd
            join ' + {{ source('adco', 'src_field') }} + N' sf on sd.src_field_id = sf.src_field_id
            join ' + {{ source('adco', 'src_mapping') }} + N' sm on sm.src_field_id = sf.src_field_id
            join ' + {{ source('adco', 'adco_field') }} + N' af on af.adco_field_id = sm.adco_field_id
        where 
            effective_date = ''2024-02-08''
    )
    select * into {{ this }}
    from source_data
    pivot (
        max(field_value) for adco_field_name in (' + @pivot_cols + N')
    ) as pvt;
';

exec sp_executesql @sql;

关键说明 #

• 两个方案均使用max(field_value)聚合函数，因Pivot必须配合聚合操作；若adco_field_name唯一，max/min/first结果一致
• 动态列名通过子查询获取去重后的adco_field_name，避免重复列
• dbt的{{ this }}变量会自动替换为当前模型的目标表名

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