PySpark基于窗口函数结合CASE语句实现客户分组时间行筛选求助

核心思路不需要复杂的嵌套CASE判断，只需要通过窗口排序时的空值位置控制，天然匹配三层筛选规则即可，之前的窗口写法出问题基本都是没有控制空值排序位置导致的。

操作步骤 #

• 第一步：将三个时间字段从字符串类型转为可正确比较的时区时间戳，避免字符串排序带来的误差
• 第二步：按customer_number做窗口分区，按照优先级规则设置排序逻辑：
  • 最高优先级：所有带非空acct_registration_ts的行排在最前面，同一类行内部按注册时间从大到小排序，空值行全部排在这类行之后
  • 次优先级：当行的注册时间为空时，带非空last_login_ts的行排在空登录时间的行前面，同一类行内部按登录时间从大到小排序
  • 最低优先级：当注册时间、登录时间都为空时，直接按acct_create_ts从大到小排序
• 第三步：给每个分组内的行计算行号，取行号为1的行即为目标结果

参考实现代码 #

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

# 1. 转换时间字段为标准时间戳类型
time_columns = ["acct_registration_ts", "last_login_ts", "acct_create_ts"]
for time_col in time_columns:
    df = df.withColumn(
        time_col,
        F.to_timestamp(time_col, "yyyy-MM-dd'T'HH:mm:ss.SSSz")
    )

# 2. 定义带优先级排序的窗口
priority_window = Window.partitionBy("customer_number").orderBy(
    # 注册时间非空行优先，时间越大越靠前，空值排最后
    F.col("acct_registration_ts").desc().nulls_last(),
    # 注册时间打平时，登录时间非空行优先，时间越大越靠前，空值排最后
    F.col("last_login_ts").desc().nulls_last(),
    # 前两列都打平时，创建时间越大越靠前
    F.col("acct_create_ts").desc().nulls_last()
)

# 3. 计算组内行号并筛选目标行
result_df = (
    df.withColumn("row_num", F.row_number().over(priority_window))
    .filter(F.col("row_num") == 1)
    .drop("row_num")
)

常见踩坑说明 #

• PySpark默认对desc排序的列会把空值排在最前面，必须显式加上.nulls_last()，否则空注册时间的行会排到非空行前面，完全违背筛选规则
• 不要直接拿原始字符串时间做排序，遇到格式偏移、补零不一致的情况会出现排序错误
• 不需要额外写多层CASE判断去统计分组内是否存在非空注册时间/登录时间的行，通过排序时空值的位置控制，就能自动实现分层筛选的效果

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