如何高效计算指定日期点的分品类滚动窗口函数（SQL/PySpark）

需要按品类维度，计算指定日期点位的过去x天（示例中x=7）指标均值，要求：

• 仅对给定的目标日期点位输出计算结果
• 目标计算日期不一定存在于明细数据的日期集合中
• 原有实现为跨表区间Join+分组聚合，大数据量下运行效率低，需要基于PySpark SQL环境用窗口函数改造实现。

涉及两张基础表：

• mydata：明细数据表
  • date：日期，TIMESTAMP类型
  • value：待统计指标值，FLOAT类型
  • category：品类标识，CHAR类型
• mypoints：待计算点位表
  • date：目标计算日期，TIMESTAMP类型
  • category：对应品类标识，CHAR类型

原有低效率实现SQL如下：

DROP TABLE IF EXISTS mydata;
DROP TABLE IF EXISTS mypoints;
CREATE TABLE mydata(
  date TIMESTAMP,
  value FLOAT,
  category CHAR
);
CREATE TABLE mypoints(
  date TIMESTAMP,
  category CHAR
);

insert into mydata VALUES
  ('2022-05-01', 7, 'a'),
  ('2022-05-01', 11, 'b'),
  ('2022-05-02', 9, 'a'),
  ('2022-05-02', 2, 'a'),
  ('2022-05-03', 42, 'b'),
  ('2022-05-04', 3, 'a'),
  ('2022-05-06', 4, 'a'),
  ('2022-05-06', 15, 'b'),
  ('2022-05-10', 1, 'a'),
  ('2022-05-10', 0, 'a');
INSERT INTO mypoints VALUES
  ('2022-05-02', 'a'),
  ('2022-05-03', 'a'),
  ('2022-05-09', 'a'),
  ('2022-05-10', 'a'),
  ('2022-05-01', 'b'),
  ('2022-05-03', 'b'),
  ('2022-05-04', 'b'),
  ('2022-05-12', 'b');
  
SELECT mp.category, mp.date, AVG(md.value)
FROM mypoints AS mp
  JOIN mydata AS md
    ON md.date BETWEEN (mp.date - INTERVAL '7 day') AND (mp.date - INTERVAL '1 day')
    AND mp.category = md.category
GROUP BY mp.date, mp.category
ORDER BY mp.category, mp.date

核心思路：避免两表区间Join产生的大量匹配开销，将待计算点位和明细数据按品类+日期合并后，使用范围滑动窗口在分区内一次性完成所有计算，最后过滤出目标点位结果即可，整体仅需一次按品类的Shuffle，计算效率远高于区间Join方案，且天然支持目标日期不在明细日期集的场景。

PySpark SQL 实现 #

WITH all_data AS (
  -- 引入明细数据，标记为非计算点位
  SELECT 
    date,
    category,
    value,
    0 AS is_calc_point
  FROM mydata
  UNION ALL
  -- 引入待计算点位，指标值置空不参与聚合，标记为目标输出点位
  SELECT
    date,
    category,
    CAST(NULL AS FLOAT) AS value,
    1 AS is_calc_point
  FROM mypoints
),
window_result AS (
  SELECT
    category,
    date,
    is_calc_point,
    -- 按品类分区，按时间戳升序，窗口范围为当前日期前7天到前1天（不含当日）
    AVG(value) OVER (
      PARTITION BY category
      ORDER BY CAST(date AS LONG)
      RANGE BETWEEN 7 * 24 * 3600 * 1000 PRECEDING AND 1 PRECEDING
    ) AS past_7d_avg_value
  FROM all_data
)
-- 仅输出目标点位的计算结果
SELECT category, date, past_7d_avg_value
FROM window_result
WHERE is_calc_point = 1
ORDER BY category, date

注意：如果date字段是DATE类型而非TIMESTAMP，无需转毫秒时间戳，将窗口边界改为RANGE BETWEEN 7 PRECEDING AND 1 PRECEDING即可。

PySpark DataFrame API 实现 #

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 给两张表打标后合并
mydata_with_tag = mydata.withColumn("is_calc_point", F.lit(0))
mypoints_with_tag = mypoints.withColumn(
    "value", F.lit(None).cast("float")
).withColumn("is_calc_point", F.lit(1))
all_data = mydata_with_tag.unionByName(mypoints_with_tag)

# 定义滑动窗口规范
win_spec = Window.partitionBy("category") \
    .orderBy(F.col("date").cast("long")) \
    .rangeBetween(-7 * 24 * 3600 * 1000, -1)  # DATE类型下偏移量改为-7到-1即可

# 计算窗口值，过滤得到最终结果
final_result = all_data.withColumn(
    "past_7d_avg_value", F.avg("value").over(win_spec)
).filter(
    F.col("is_calc_point") == 1
).select(
    "category", "date", "past_7d_avg_value"
).orderBy("category", "date")

final_result.show()

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