如何使用Presto SQL查询含结转库存的各商品每日最高库存数

需求核心说明 #

我们需要统计每个商品每日的最高库存，规则为：

• 当日有库存变动记录时，取「当日所有库存记录的最大值」和「前一日结转的库存值」两者的较大值
• 当日无库存变动记录时，最高库存直接等于前一日结转的库存值

完整实现SQL #

WITH 
-- 步骤1：基础聚合，计算每个商品每日的最大库存、日终库存
daily_agg AS (
    SELECT 
        date_format(from_unixtime(CAST(update_timestamp AS BIGINT)/1000), '%Y-%m-%d') AS day,
        item_id,
        MAX(units) AS daily_max_units,
        -- 用MAX_BY取当日最后一笔时间戳对应的库存，即当日结转至次日的库存
        MAX_BY(units, update_timestamp) AS daily_end_units
    FROM your_table_name -- 替换为你的实际表名
    GROUP BY item_id, date_format(from_unixtime(CAST(update_timestamp AS BIGINT)/1000), '%Y-%m-%d')
),
-- 步骤2：生成业务时间范围内的连续日期序列
date_range AS (
    SELECT 
        date_format(day_seq, '%Y-%m-%d') AS day
    FROM (
        SELECT UNNEST(SEQUENCE(
            -- 时间范围可以根据业务需要自定义，这里默认取表中最早到最晚的事务日期
            (SELECT MIN(date(from_unixtime(CAST(update_timestamp AS BIGINT)/1000))) FROM your_table_name),
            (SELECT MAX(date(from_unixtime(CAST(update_timestamp AS BIGINT)/1000))) FROM your_table_name),
            INTERVAL '1' DAY
        )) AS day_seq
    )
),
-- 步骤3：生成每个商品对应的全量日期序列，补全无事务的空日期
item_full_dates AS (
    SELECT 
        dr.day,
        i.item_id
    FROM date_range dr
    CROSS JOIN (SELECT DISTINCT item_id FROM your_table_name) i
),
-- 步骤4：关联聚合数据，用窗口函数填充结转值
joined_data AS (
    SELECT 
        ifd.day,
        ifd.item_id,
        da.daily_max_units,
        da.daily_end_units,
        -- 前向填充最近的非空日终库存，作为当日的结转基础
        LAST_VALUE(da.daily_end_units IGNORE NULLS) OVER (
            PARTITION BY ifd.item_id 
            ORDER BY ifd.day 
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) AS carry_over_end_units,
        -- 取上一日的结转库存，用于和当日最大值比较
        LAG(LAST_VALUE(da.daily_end_units IGNORE NULLS) OVER (
            PARTITION BY ifd.item_id 
            ORDER BY ifd.day 
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ), 1) OVER (PARTITION BY ifd.item_id ORDER BY ifd.day) AS prev_day_carry_over
    FROM item_full_dates ifd
    LEFT JOIN daily_agg da ON ifd.day = da.day AND ifd.item_id = da.item_id
)
-- 最终计算每日最高库存
SELECT 
    day,
    item_id,
    COALESCE(GREATEST(daily_max_units, prev_day_carry_over), daily_max_units, carry_over_end_units) AS max_units
FROM joined_data
ORDER BY item_id, day

兼容说明 #

如果你的Presto版本不支持MAX_BY函数，可以将daily_agg部分替换为以下ROW_NUMBER实现：

daily_agg AS (
    WITH daily_rn AS (
        SELECT 
            date_format(from_unixtime(CAST(update_timestamp AS BIGINT)/1000), '%Y-%m-%d') AS day,
            item_id,
            units,
            ROW_NUMBER() OVER (PARTITION BY item_id, date_format(from_unixtime(CAST(update_timestamp AS BIGINT)/1000), '%Y-%m-%d') ORDER BY update_timestamp DESC) AS rn
        FROM your_table_name
    )
    SELECT 
        day,
        item_id,
        MAX(units) AS daily_max_units,
        MAX(CASE WHEN rn = 1 THEN units END) AS daily_end_units
    FROM daily_rn
    GROUP BY item_id, day
)

效果验证 #

以你给出的示例为例：

• item1 2021-11-24的日终库存为6，2021-11-25无事务，当日最高库存取结转值6
• item1 2021-11-26的当日最大库存为4，和前一日结转值6取较大值，最终最高库存为6，完全符合需求。

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