PostgreSQL/DBT实现JSONB列扁平化及数据表规范化求助

前提说明 #

从你的SQL查询来看，additionalFields是JSONB数组类型，每个数组元素是包含name和对应值的对象（比如{"name":"Open Cases", "value":"..."}）。下面是用dbt实现规范化的具体步骤：

步骤1：创建Staging模型（获取原始数据） #

在dbt项目的models/staging目录下创建stg_tasks.sql，读取原始tasks表的数据：

WITH source AS (
    SELECT
        "id",
        "additionalFields"
    FROM {{ source('your_source_schema', 'tasks') }}
    -- 替换your_source_schema为你实际的源数据schema（比如public）
    WHERE "additionalFields" @> '[{"name":"Open Cases"}]' -- 保留你的过滤条件
)

SELECT * FROM source

步骤2：拆分JSONB数组并提取键值对 #

在models/intermediate目录创建int_tasks_flattened.sql，把数组拆分成单独的键值对行：

WITH stg AS (
    SELECT * FROM {{ ref('stg_tasks') }}
),

flattened AS (
    SELECT
        "id",
        -- 提取数组对象的name作为字段名，value作为对应值
        (jsonb_array_elements("additionalFields"))->>'name' AS field_name,
        (jsonb_array_elements("additionalFields"))->>'value' AS field_value
        -- 若值为数字/布尔类型，可用::int/::boolean强转类型
    FROM stg
)

SELECT * FROM flattened

步骤3：Pivot行转列生成规范化表 #

在models/marts目录创建最终模型tasks_normalized.sql，根据字段是否固定选择对应方式：

情况1：字段名固定（已知所有需要的字段） #

如果additionalFields里的name是固定的（比如仅"Open Cases"、"Priority"等），直接写死列逻辑：

WITH flattened AS (
    SELECT * FROM {{ ref('int_tasks_flattened') }}
)

SELECT
    "id",
    MAX(CASE WHEN field_name = 'Open Cases' THEN field_value END) AS open_cases,
    MAX(CASE WHEN field_name = 'Priority' THEN field_value END) AS priority,
    MAX(CASE WHEN field_name = 'Assignee' THEN field_value END) AS assignee
    -- 按需添加其他字段
FROM flattened
GROUP BY "id"

情况2：字段名动态变化 #

如果additionalFields里的name不固定，用dbt宏实现动态pivot：

1. 在macros/dynamic_pivot.sql创建宏：

{% macro dynamic_pivot(model, pivot_column, value_column, group_by_column) %}
    {% set query %}
        SELECT DISTINCT {{ pivot_column }} FROM {{ model }}
    {% endset %}

    {% set results = run_query(query) %}

    {% if execute %}
        {% set pivot_values = results.columns[0].values() %}
    {% else %}
        {% set pivot_values = [] %}
    {% endif %}

    SELECT
        {{ group_by_column }},
        {% for value in pivot_values %}
            MAX(CASE WHEN {{ pivot_column }} = '{{ value }}' THEN {{ value_column }} END) AS {{ value | lower | replace(' ', '_') }}
            {% if not loop.last %},{% endif %}
        {% endfor %}
    FROM {{ model }}
    GROUP BY {{ group_by_column }}
{% endmacro %}

2. 在最终模型中调用宏：

WITH flattened AS (
    SELECT * FROM {{ ref('int_tasks_flattened') }}
)

{{ dynamic_pivot('flattened', 'field_name', 'field_value', 'id') }}

运行dbt #

完成模型编写后，执行命令生成规范化表：

dbt run --models tasks_normalized

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