PySpark带过滤的Window函数应用：基于指定DataFrame实现数据提取

看来你已经搭好了Spark窗口分析的基础框架，我给你整理几个常用的数据提取场景，直接上手就能用：

1. 获取每个id的首/末次记录 #

如果需要拿到每个用户（按id区分）最早和最晚时间点的记录，可以这么实现：

from pyspark.sql import functions as F

# 提取每个id的第一条（最早时间）记录
first_records = df.withColumn("is_first", F.row_number().over(w) == 1) \
                  .filter(F.col("is_first")) \
                  .drop("is_first")

# 提取每个id的最后一条（最晚时间）记录
last_records = df.withColumn("row_num", F.row_number().over(w)) \
                 .withColumn("max_row", F.max("row_num").over(Window.partitionBy("id"))) \
                 .filter(F.col("row_num") == F.col("max_row")) \
                 .drop("row_num", "max_row")

2. 计算相邻记录的x值变化 #

想知道每个id每次记录的x相对于上一条的增减量？用lag()函数就能轻松实现：

# 计算当前x与上一条记录x的差值
x_diff_df = df.withColumn("prev_x", F.lag("x").over(w)) \
              .withColumn("x_change", F.col("x") - F.col("prev_x"))

3. 捕捉y值从0到1的切换记录 #

如果你的需求是找到每个id中y第一次从0变为1的那条记录，可以结合窗口函数和条件筛选：

# 标记上一条记录的y值，筛选y从0转1的过渡记录
y_transition_df = df.withColumn("prev_y", F.lag("y").over(w)) \
                    .filter((F.col("y") == 1) & (F.col("prev_y") == 0))

4. 滚动统计（比如最近2条x的平均值） #

要做基于时间顺序的滚动计算，比如每个id当前及前一条记录的x平均值，可以用窗口的行数范围参数：

# 计算每个id当前及上一条记录的x平均值
rolling_avg_df = df.withColumn("rolling_x_avg", F.avg("x").over(w.rowsBetween(-1, 0)))

如果你有更具体的提取需求（比如特定条件过滤、聚合计算等），可以补充说明细节，我再帮你调整代码~

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