本文详解如何在 Polars 中高效筛选满足「同一用户下，当前记录之后（含/不含自身）5 分钟内存在类别为 'A' 的事件」的原始行，避免 SQL .sql() 方法，提供可复用的 DataFrame 级滚动窗口+列表判断方案，并阐明 SQL 窗口函数到 Polars 的核心映射逻辑。
本文详解如何在 polars 中高效筛选满足「同一用户下，当前记录之后（含/不含自身）5 分钟内存在类别为 'a' 的事件」的原始行，避免 sql `.sql()` 方法，提供可复用的 dataframe 级滚动窗口+列表判断方案，并阐明 sql 窗口函数到 polars 的核心映射逻辑。
在 Polars 中实现类似 SQL COUNT_IF(...) OVER (PARTITION BY user_id ORDER BY t RANGE BETWEEN CURRENT ROW AND INTERVAL '5 MINUTES' FOLLOWING) 的逻辑，关键在于理解：Polars 的表达式级 rolling 不支持 over + 时间范围窗口组合，必须使用 DataFrame 级 .rolling(..., group_by=...)，再通过聚合保留原始行信息并做条件过滤。
✅ 正确思路：DataFrame 滚动窗口 + 聚合保行 + 列表判断
核心步骤如下：
-
添加行索引：用于最终恢复原始顺序（
.rolling会打乱顺序）； -
按
user_id分组、以时间列t为索引进行滚动窗口聚合：设置period="5m"、closed="both"（含当前行），覆盖未来 5 分钟； -
聚合时保留原始行首值（
pl.all().first()）并额外提取cat列为列表（pl.col("cat").alias("_filt")）； -
过滤
_filt列中是否包含"A"（list.contains("A")）； - 按原始索引排序并清理辅助列。
以下是完整可运行代码：
import polars as pl
from datetime import datetime
# 示例数据
df = pl.DataFrame(
[
pl.Series("t", [
datetime(2025, 1, 1, 0, 0),
datetime(2025, 1, 1, 0, 0),
datetime(2025, 1, 1, 0, 0),
datetime(2025, 1, 1, 0, 1),
datetime(2025, 1, 1, 0, 3),
datetime(2025, 1, 1, 0, 3),
datetime(2025, 1, 1, 0, 4),
datetime(2025, 1, 1, 0, 5),
datetime(2025, 1, 1, 0, 5),
datetime(2025, 1, 1, 0, 6)
], dtype=pl.Datetime),
pl.Series("user_id", ["Bart", "Aaron", "Charles", "Aaron", "Aaron", "Charles", "Bart", "Bart", "Aaron", "Aaron"]),
pl.Series("cat", ["C", "C", "B", "A", "C", "B", "A", "A", "B", "A"]),
]
)
# ✅ 核心实现：筛选「当前行起 5 分钟内（含自身）存在 'A'」的记录
result = (
df.with_row_index("i") # 步骤1：添加索引保序
.rolling(
index_column="t",
period="5m",
offset="0m",
group_by="user_id",
closed="both" # 包含当前行及之后5分钟内所有行
)
.agg(
pl.all().first(), # 保留每组窗口首行的原始各列值
pl.col("cat").alias("_filt") # 同时提取该窗口内所有 cat 构成列表
)
.filter(pl.col("_filt").list.contains("A")) # 步骤4：判断列表中是否有 'A'
.sort("i") # 步骤5：按原始顺序恢复
.drop(["_filt", "i"]) # 清理辅助列
)
print(result)
? 输出说明：结果包含 8 行，例如
Bart@00:00被保留，因其后续00:04和00:05均为'A'；而Charles@00:03未被保留，因该窗口内仅["B"]，无'A'。
⚠️ 若需排除当前行（仅看“后续”5分钟）
SQL 中 RANGE BETWEEN CURRENT ROW AND ... 默认含当前行；若需求是「严格后续」（即不包含当前行本身），需对 _filt 列切片后再判断：
.filter(pl.col("_filt").list.slice(1, None).list.contains("A"))
list.slice(1, None) 表示跳过第一个元素（即当前行），只检查后续事件。修改后结果为 6 行，例如 Bart@00:04 和 Bart@00:05 因自身为 'A' 但后续无 'A'（窗口内仅 ["A"] → 切片后为空 → 不满足），故被排除。
? SQL → Polars 窗口函数心智模型对照表
| SQL 概念 | Polars 等价实现 | 注意事项 |
|---|---|---|
PARTITION BY X ORDER BY Y |
.rolling(..., group_by="X", index_column="Y") |
必须用 DataFrame 级 .rolling，不能在 Expr 上链式调用 .over()
|
RANGE BETWEEN CURRENT ROW AND INTERVAL '5m' FOLLOWING |
period="5m", closed="both"（默认含当前）或 closed="right"（仅后续） |
closed="left"=仅当前及之前；"right"=仅之后；"both"=当前+前后；"none"=仅当前 |
COUNT_IF(cat='A') > 0 |
pl.col("cat").list.contains("A") |
先聚合出列表，再用 list.* 方法判断 |
| 保持原始粒度（不降行） |
pl.all().first() + 行索引保序 |
.agg() 必然聚合，需用 first() 还原单行，并靠索引恢复顺序 |
✅ 总结
- Polars 不支持 SQL 风格的“表达式级时间范围窗口”，必须转向 DataFrame 滚动聚合 + 列表操作 范式；
- 关键技巧：用
pl.all().first()锁定窗口内首行原始数据，用pl.col("cat").list.*对窗口内值做集合运算； - 务必插入
.with_row_index()并.sort()，否则结果顺序不可控； -
closed参数决定是否包含边界行，list.slice()可灵活控制是否纳入当前行。
此模式可轻松泛化至任意事件类型（如 'B', 'login'）、任意时间窗口（"10s", "1h"）及多条件组合（如 list.contains("A") & list.contains("B")），是 Polars 处理时序依赖筛选的稳健实践路径。
