如何用PySpark/Snowpark按ID计算自定义日期区间销售累计和

现有销售数据表如下：

需要计算每个ID在自定义日期区间内的销售总额：

• ID 1：2020-04-30 至 2021-04-30
• ID 2：2020-08-31 至 2021-08-31

尝试用窗口函数但无法适配每个ID的自定义区间，现有代码片段：

win = Window.partitionBy('id').orderBy('sales_date')
df.withColumn('running_sum',sum(sales).over(win).rangeBetween(start_date,end_date) ??

# rangeBetween of start_date and start_date + 1 year

方法1：关联区间映射表后分组求和 #

这种方法逻辑直观，完全适配任意自定义区间的场景：

1. 准备数据并转换日期格式：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# 初始化SparkSession
spark = SparkSession.builder.appName("custom_date_range_sum").getOrCreate()

# 加载原销售数据并转换日期类型
sales_data = [
    (1, 10, "2020-04-30"),
    (1, 6, "2020-10-31"),
    (1, 9, "2020-09-30"),
    (1, 2, "2021-04-30"),
    (2, 8, "2020-08-31"),
    (2, 7, "2020-07-31"),
    (2, 3, "2021-06-30"),
    (2, 2, "2021-05-31")
]
sales_df = spark.createDataFrame(sales_data, ["id", "sales", "sales_date"])
sales_df = sales_df.withColumn("sales_date", col("sales_date").cast("date"))

# 创建ID-自定义区间映射表并转换日期类型
range_data = [
    (1, "2020-04-30", "2021-04-30"),
    (2, "2020-08-31", "2021-08-31")
]
range_df = spark.createDataFrame(range_data, ["id", "start_date", "end_date"])
range_df = range_df.withColumn("start_date", col("start_date").cast("date")) \
                   .withColumn("end_date", col("end_date").cast("date"))

2. 关联过滤并计算总额：

# 关联表，过滤出每个ID区间内的销售记录
filtered_df = sales_df.join(range_df, on="id", how="inner") \
                      .where(col("sales_date").between(col("start_date"), col("end_date")))

# 按ID分组求和
result_df = filtered_df.groupBy("id") \
                       .agg({"sales": "sum"}) \
                       .withColumnRenamed("sum(sales)", "total_sales")

# 查看结果
result_df.show()

执行结果：

+---+-----------+
| id|total_sales|
+---+-----------+
|  1|         27|
|  2|         13|
+---+-----------+

方法2：窗口函数结合条件求和（适合区间有规律的场景） #

如果区间是类似"起始日期+1年"这类有固定规则的场景，可以用窗口函数实现，但无规律自定义区间推荐用方法1：

from pyspark.sql.window import Window
from pyspark.sql.functions import sum, when, min, add_months

# 用窗口函数获取每个ID的最早销售日期作为起始，加1年作为结束日期（示例规则）
win = Window.partitionBy("id")
df_with_range = sales_df.withColumn("start_date", min("sales_date").over(win)) \
                        .withColumn("end_date", add_months(col("start_date"), 12))

# 条件求和计算区间内总额
result_df = df_with_range.groupBy("id") \
                         .agg(sum(when(col("sales_date").between(col("start_date"), col("end_date")), col("sales"))).alias("total_sales"))

result_df.show()

关键说明 #

• 窗口函数的rangeBetween多用于基于数值偏移的范围（如前后N天），但针对每个ID不同的固定区间，关联映射表+过滤+分组求和的逻辑更简单直接，易维护。
• 方法1的映射表可以灵活扩展，支持任意ID的任意日期区间配置。

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