如何在PySpark窗口函数rangeBetween中用列值计算最大延迟天数

你遇到的报错是因为rangeBetween的参数使用错误：

1. 当使用降序排序（tax_period_ts.desc()）时，rangeBetween的起始值必须小于等于结束值，但你传入的as_on_date_ts（较大时间戳）在前，days_90_before_u_ts（较小时间戳）在后，范围逻辑反向，导致Spark无法解析列的布尔判断。
2. rangeBetween的参数顺序需要与orderBy的排序方向匹配：升序排序时，范围是「较小值 → 较大值」；降序排序时则相反，这很容易混淆。

利用数据按company_master_id和as_on_date分区的特性（每个分区内as_on_date固定），将日期转换为时间戳后，基于升序排序的t_tax_period时间戳定义窗口范围，直接计算指定天数内的max(delay_days)。

完整代码示例 #

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

# 1. 将日期字段转换为Unix时间戳（秒）
df = df.withColumn("t_tax_period_ts", F.unix_timestamp(F.col("t_tax_period"))) \
       .withColumn("as_on_date_ts", F.unix_timestamp(F.col("as_on_date")))

# 2. 定义通用函数批量生成不同天数的最大延迟值
def add_max_delay_column(df, days):
    # 定义窗口：按公司和日期分区，按税期时间戳升序，范围为as_on_date往前推days天到as_on_date当天
    window_spec = Window.partitionBy("company_master_id", "as_on_date") \
                       .orderBy("t_tax_period_ts") \
                       .rangeBetween(
                           F.col("as_on_date_ts") - days * 86400,  # 往前推days天的时间戳（86400秒=1天）
                           F.col("as_on_date_ts")                  # as_on_date当天的时间戳
                       )
    col_name = f"{days}_days_max_delay"
    return df.withColumn(col_name, F.max("delay_days").over(window_spec))

# 3. 计算90/180/365天内的最大延迟值
df = add_max_delay_column(df, 90)
df = add_max_delay_column(df, 180)
df = add_max_delay_column(df, 365)

# 可选：清理临时生成的时间戳列
df = df.drop("t_tax_period_ts", "as_on_date_ts")

关键说明 #

• 改用升序排序的t_tax_period_ts，符合时间从早到晚的逻辑，避免范围顺序混淆。
• 直接通过as_on_date_ts - days*86400计算时间范围下界，无需依赖预先生成的days_90_before列，简化逻辑。
• 通用函数add_max_delay_column减少重复代码，便于扩展其他天数的计算需求。

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