RangeBetween与RowsBetween选择及时间感知窗口累积聚合问题

一、针对时间感知累积聚合的选择 #

两者核心区别在于窗口范围的界定逻辑：

• rowsBetween：按行的物理顺序/数量确定窗口，比如rowsBetween(UNBOUNDED_PRECEDING, CURRENT_ROW)会包含当前分区内从第一行到当前行的所有行，和行的内容无关。适合不需要基于值范围的行顺序累积场景。
• rangeBetween：按排序后的值范围确定窗口，比如按日期排序后，rangeBetween(UNBOUNDED_PRECEDING, CURRENT_ROW)会包含当前分区内所有值<=当前行排序值的行。这完全匹配你的时间感知需求——同日期的所有行都会被纳入累积计算。

针对你提供的分区"c"数据（所有日期相同）：

• 若用rangeBetween且按insert_date排序，分区内所有行的日期都等于当前行日期，因此每一行的累积求和结果都是350+100+10=460。
• 若用rowsBetween，则会按行的输入顺序依次累积，结果为350、450、460，不符合时间感知的需求。

二、rangeBetween的使用规则 #

1. 必须指定orderBy：range是基于排序后的值范围计算的，无排序则无法确定范围，这也是你代码报错的直接原因——你的窗口定义没有添加orderBy就使用了rangeBetween。
2. 排序列支持非数值类型：日期、时间戳类型都可以作为排序列，Spark会自动将其转换为对应的数值（比如日期转为自1970-01-01的天数）来计算范围。

三、让order_col可选的实现方案 #

如果要将order_col设为可选参数，需分场景处理：

• 当指定order_col时：根据需求选择rangeBetween（时间/值感知）或rowsBetween（行顺序感知）。
• 当未指定order_col时：只能使用rowsBetween，但此时累积顺序依赖数据的物理存储，结果不可控，建议在函数中添加校验——若使用rangeBetween则强制要求传入order_col。

四、修正后的代码实现 #

你的代码存在三个核心问题：窗口未加orderBy、循环变量未定义、函数调用时未传必填的order_col。以下是修正后的完整代码：

from pyspark.sql import DataFrame, Window, functions as F
from pyspark.sql.types import StructType, StructField, StringType, DateType, IntegerType
from typing import Dict, Union, List
from datetime import date

def get_cumulative_agg_by_window(
    df: DataFrame,
    agg_dict: Dict[str, List[str]],
    partition_cols: Union[str, List[str]],
    order_col: str = None,
    use_range: bool = True
) -> DataFrame:
    """提供窗口内的累积聚合，支持自定义列名，可选时间/值感知或行顺序感知模式

    Args:
        df (DataFrame): 输入DataFrame
        agg_dict (Dict[str, List[str]]): 聚合配置，键为待聚合列名，值为聚合方法列表（如["min", "max", "sum", "mean", "first"]）
        partition_cols (Union[str, List[str]]): 分区列
        order_col (str, optional): 排序列，使用rangeBetween时必填
        use_range (bool, optional): 是否使用rangeBetween（时间/值感知），默认True
    """
    # 基础窗口定义
    window_spec = Window.partitionBy(partition_cols)

    # 校验参数：使用rangeBetween时必须传order_col
    if use_range and not order_col:
        raise ValueError("order_col is required when use_range is True")

    # 构建窗口范围
    if order_col:
        window_spec = window_spec.orderBy(order_col)
        window_frame = window_spec.rangeBetween(Window.unboundedPreceding, Window.currentRow) if use_range else window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow)
    else:
        # 无排序列时，只能用rowsBetween，累积顺序依赖物理存储
        window_frame = window_spec.rowsBetween(Window.unboundedPreceding, Window.currentRow)

    # 执行聚合
    for col_name, aggs in agg_dict.items():
        for agg in aggs:
            # 兼容avg别名（Spark中对应mean）
            agg_func = getattr(F, "mean" if agg == "avg" else agg)
            new_col = f"cumulative_{agg}_{col_name}"
            df = df.withColumn(new_col, agg_func(F.col(col_name)).over(window_frame))

    return df

# 测试数据准备
schema = StructType([
    StructField("partition_col", StringType(), True),
    StructField("insert_date", DateType(), True),
    StructField("item", StringType(), True),
    StructField("sales", IntegerType(), True),
])

data = [
    ("A", date(2021, 6, 1), "apple", 100),
    ("A", date(2022, 6, 2), "banana", 150),
    ("A", date(2022, 6, 3), "banana", 200),
    ("B", date(2023, 6, 1), "apple", 250),
    ("B", date(2023, 6, 2), "apple", 300),
    ("B", date(2022, 6, 3), "banana", 350),
    ("c", date(2022, 6, 3), "banana", 350),
    ("c", date(2022, 6, 3), "apple", 100),
    ("c", date(2022, 6, 3), "banana", 10),
]

test_df = spark.createDataFrame(data, schema)

# 测试调用：按partition_col分区，insert_date排序，range模式计算sales的sum
agg_dict = {"sales": ["sum"]}
result_df = get_cumulative_agg_by_window(
    test_df,
    agg_dict,
    partition_cols=["partition_col"],
    order_col="insert_date",
    use_range=True
)

result_df.orderBy("partition_col", "insert_date").show()

五、测试结果说明 #

运行上述代码后，分区"c"的每一行cumulative_sum_sales值都是460，符合时间感知的累积需求；若将use_range设为False，则分区"c"的结果会按行顺序依次显示350、450、460。

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