Trino DB中动态清洗JSON历史数据并转换为列的实现方案

数据库中progress_history列存储JSON字符串，示例如下：

eg1: {'2022-11-01': 44038.91099579527, '2022-12-05': 44038.91099579527, '2023-01-21': 20623.839917193636, '2023-02-01': 20623.839917193636, '2023-02-25': 20623.839917193636}
eg2: {'2023-05-26T12:53:50.708195+00:00': 0.0, '2023-05-26T12:53:51.034034+00:00': 0}

需求：获取当前日期过去6个月的每个月对应的指标值，规则为取当月最早时间点的数值（如eg2中2023年5月取最早时间点的0.0）。

原尝试代码仅能获取每月1日的数值，且无法动态生成月份列。

核心思路 #

先将JSON键值对展开为行数据，统一处理时间格式后筛选目标时间段，再通过窗口函数提取每月最早时间点的数值，最后可选转置为月份列。

完整Trino代码 #

WITH parsed_data AS (
    -- 解析JSON并展开键值对
    SELECT
        k.kpi,
        -- 兼容两种时间键格式，转为日期类型
        CASE
            WHEN regexp_like(hist.key, '^\d{4}-\d{2}-\d{2}$') THEN date(hist.key)
            ELSE date(hist.key)
        END AS record_date,
        hist.value AS metric_value,
        -- 转为时间戳用于排序取最早值
        CASE
            WHEN regexp_like(hist.key, '^\d{4}-\d{2}-\d{2}$') THEN timestamp(hist.key || 'T00:00:00')
            ELSE timestamp(hist.key)
        END AS record_timestamp
    FROM
        kpis k,
        UNNEST(map_entries(json_parse(replace(k.progress_history, '''', '"')))) AS hist(key, value)
),
filtered_data AS (
    -- 筛选过去6个月的记录
    SELECT
        kpi,
        metric_value,
        record_timestamp,
        date_trunc('month', record_date) AS record_month
    FROM parsed_data
    WHERE record_month >= date_trunc('month', current_date - interval '6' month)
),
monthly_metrics AS (
    -- 提取每个月最早时间点的数值
    SELECT
        kpi,
        record_month,
        first_value(metric_value) OVER (
            PARTITION BY kpi, record_month
            ORDER BY record_timestamp ASC
            ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
        ) AS monthly_value
    FROM filtered_data
    GROUP BY kpi, record_month, metric_value, record_timestamp
),
deduped_metrics AS (
    -- 去重，确保每个kpi+月份仅保留一条数据
    SELECT DISTINCT kpi, record_month, monthly_value
    FROM monthly_metrics
)
-- 动态转置为月份列（若需行式结果，直接查询deduped_metrics即可）
SELECT *
FROM deduped_metrics
PIVOT (
    MAX(monthly_value)
    FOR record_month IN (
        date_trunc('month', current_date - interval '5' month),
        date_trunc('month', current_date - interval '4' month),
        date_trunc('month', current_date - interval '3' month),
        date_trunc('month', current_date - interval '2' month),
        date_trunc('month', current_date - interval '1' month),
        date_trunc('month', current_date)
    )
) AS p(kpi, prev_5_month, prev_4_month, prev_3_month, prev_2_month, prev_1_month, current_month);

代码说明 #

1. JSON解析与展开：通过json_parse将清洗后的JSON转为Map类型，再用UNNEST(map_entries(...))把键值对拆分为行，便于逐条处理时间和数值。
2. 时间格式兼容：用CASE语句处理两种时间键格式，统一转为date和timestamp，确保排序和筛选逻辑正确。
3. 时间段筛选：通过current_date - interval '6' month动态计算过去6个月的起始点，避免硬编码。
4. 取最早数值：使用first_value窗口函数，按时间戳升序提取每个月的第一个数值，保证拿到当月最早记录。
5. 动态列生成：PIVOT中的月份通过current_date动态计算，每次执行都会自动适配最新的过去6个月区间。

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