本文详解如何在 pyspark 中构建复合窗口,同时满足“最近 n 行”和“最近 m 天”双重约束,实现带结构化历史记录的列计算。
在实际数据分析场景中,仅依赖固定行偏移(如 rowsBetween(-2, -1))的窗口往往无法准确刻画业务逻辑——例如用户行为分析常要求:“取当前行前最多 2 条、且日期不超过 10 天的记录”。PySpark 原生窗口不支持直接混合时间范围与行数限制,但可通过两阶段策略优雅解决:先用行窗口粗筛候选集,再用 filter() + SQL 表达式精筛时间条件。
核心实现思路
- 结构化封装:使用 struct() 将目标字段(id, date, value)打包为嵌套结构体,便于整体收集与过滤;
- 行窗口预聚合:通过 collect_list().over(Window.rowsBetween(-n, -1)) 获取最近 n 行(建议设为略大于上限,如 -3 对应“最多 2 行”);
- 时间条件后过滤:利用 expr(“filter(history, x -> x.date >= (date – interval 10 days))”) 动态剔除超时记录;
- 结果格式化(可选):若需字符串形式(如 (id, date, value)),可追加 concat_ws() 和 transform()。
以下为完整可运行示例(适配您需求的 10 天 + 最多 2 行):