Snowflake中Variant JSON列的扁平化、类型转换与动态转置问询

问题背景 #

有一张包含id列和variant类型JSON列的表，需完成以下操作：

• 对JSON列执行lateral flatten扁平化处理
• 将扁平化后的value列转为对应的数据类型（布尔、小数、字符串等）后存为variant
• 最终转置为宽表，让每个字段（如field1、field2）匹配正确的数据类型
  实际场景涉及数百个字段，希望借助dbt_utils.get_column_values动态获取列名，无需手动枚举所有字段。

用户提供的无法正常运行的示例代码：

with cte as (
    select 
        1 as id,
        parse_json('{
            "field1":"TRUE",
            "field2":"some string",
            "field3":"1.035",
            "field4":"097334"
        }') as my_output
)
select 
    id,
    key,
    to_variant(
        case
            when value in ('true', 'false') then value::boolean
            when value like ('1.0') then value::decimal
            else value::string
        end
    ) as value
from cte, lateral flatten(my_output)

可行解决方案 #

完全可以实现动态处理，结合dbt宏与Snowflake特性，步骤如下：

1. 统一类型转换逻辑 #

先创建宏替换示例中逻辑受限的case语句（原代码仅匹配固定格式小数，覆盖不全）：

{% macro convert_to_correct_type(value_str) %}
    case
        -- 匹配大小写不敏感的布尔值
        when upper({{ value_str }}) in ('TRUE', 'FALSE') then {{ value_str }}::boolean::variant
        -- 匹配整数、正负小数的通用格式
        when regexp_like({{ value_str }}, '^-?\d+(\.\d+)?$') then {{ value_str }}::decimal::variant
        -- 其他情况保留字符串类型
        else {{ value_str }}::string::variant
    end
{% endmacro %}

2. 动态获取所有字段名 #

通过dbt_utils.get_column_values或直接查询提取扁平化后的所有key值（即字段名）：

-- 方式1：从已有的扁平化模型中提取
{% set field_names = dbt_utils.get_column_values(
    table=ref('your_flattened_temp_model'),
    column='key'
) %}

-- 方式2：直接查询源表获取
{% set field_names_query %}
    select distinct key from {{ ref('your_source_table') }}, lateral flatten(my_output)
{% endset %}
{% set field_names = run_query(field_names_query).columns[0].values() %}

3. 动态生成转置SQL #

用dbt循环语法自动生成转置逻辑，同时将variant转回对应数据类型：

with flattened_data as (
    select 
        id,
        key,
        {{ convert_to_correct_type('value') }} as typed_value
    from {{ ref('your_source_table') }}, lateral flatten(my_output)
),

transposed as (
    select 
        id,
        {% for field in field_names %}
            -- 用try_cast避免转换失败报错，也可根据字段名自定义类型
            try_cast(max(case when key = '{{ field }}' then typed_value end) as string) as {{ field }}
            {% if not loop.last %},{% endif %}
        {% endfor %}
    from flattened_data
    group by id
)

select * from transposed

4. 可选：精准类型映射（按需使用） #

如果需要为特定字段指定固定类型，可添加类型映射宏：

{% macro get_target_type(field_name) %}
    case
        when '{{ field_name }}' = 'field1' then 'boolean'
        when '{{ field_name }}' = 'field3' then 'decimal(10,3)'
        else 'string'
    end
{% endmacro %}

将转置逻辑中的string替换为{{ get_target_type(field) }}即可。

关键注意事项 #

• 原示例的value like ('1.0')逻辑覆盖范围极窄，改用正则表达式匹配数字格式更可靠
• 动态获取字段名时，需确保源表包含所有可能的字段，或定期刷新字段列表
• 转置时用max()/min()聚合，是因为每个id+key仅对应一个值，聚合不会改变结果
• 使用try_cast可避免单个字段转换失败导致整个查询报错

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