PySpark实现按用户判断历史是否存在Y值（向前填充逻辑）

要实现这个需求，核心是要先按用户+日期聚合，确保同一日期的所有记录视为一个整体，再通过窗口函数判断该日期之前是否出现过Y，最后关联回原表得到所有记录的结果。直接对原表用partitionBy会导致同一日期内的记录互相干扰，这应该是你之前失败的原因。

步骤分解与代码实现 #

1. 导入依赖并准备示例数据

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

# 初始化SparkSession
spark = SparkSession.builder.appName("any_previous_y").getOrCreate()

# 创建示例DataFrame（可替换为你的实际数据）
data = [
    (1, "2023-01-01", "N"),
    (1, "2023-01-01", "N"),
    (1, "2023-01-02", "Y"),
    (1, "2023-01-03", "N"),
    (2, "2023-01-01", "Y"),
    (2, "2023-01-02", "N"),
    (2, "2023-01-02", "N"),
    (3, "2023-01-01", "N"),
    (3, "2023-01-02", "N"),
    (3, "2023-01-03", "N")
]

df = spark.createDataFrame(data, ["user_id", "date", "y_n"])
# 确保date列是日期类型（如果你的数据已经是日期类型可跳过）
df = df.withColumn("date", F.to_date("date"))

2. 按用户+日期聚合，标记当日是否存在Y

这一步是为了把同一用户同一日期的所有记录合并为一条，标记该日期是否有Y（避免同一日期内的记录互相影响）：

daily_summary = df.groupBy("user_id", "date")\
    .agg(F.max("y_n").alias("has_y_today"))
# 注：字符串比较中'Y' > 'N'，所以max()会自动提取当日是否有Y

3. 用窗口函数计算每个日期之前是否出现过Y

定义窗口按用户分区、日期排序，取当前日期之前的所有日期的has_y_today最大值，以此判断是否有过Y：

# 窗口规则：按用户分区，按日期排序，只包含当前行之前的所有行（不包括当前行）
window_spec = Window.partitionBy("user_id").orderBy("date")\
    .rowsBetween(Window.unboundedPreceding, Window.preceding(1))

# 计算any_previous_y：之前有Y则为Y，否则为N（第一条记录无前驱，用coalesce处理null）
daily_summary = daily_summary.withColumn(
    "any_previous_y",
    F.coalesce(F.max("has_y_today").over(window_spec), F.lit("N"))
)

4. 关联回原表，得到最终结果

将聚合计算后的结果关联回原DataFrame，确保同一日期的所有记录都获得相同的any_previous_y值：

final_df = df.join(
    daily_summary.select("user_id", "date", "any_previous_y"),
    on=["user_id", "date"],
    how="left"
)

查看结果

final_df.show()

输出结果：

+-------+----------+----+--------------+
|user_id|      date|y_n |any_previous_y|
+-------+----------+----+--------------+
|      1|2023-01-01|   N|             N|
|      1|2023-01-01|   N|             N|
|      1|2023-01-02|   Y|             N|
|      1|2023-01-03|   N|             Y|
|      2|2023-01-01|   Y|             N|
|      2|2023-01-02|   N|             Y|
|      2|2023-01-02|   N|             Y|
|      3|2023-01-01|   N|             N|
|      3|2023-01-02|   N|             N|
|      3|2023-01-03|   N|             N|
+-------+----------+----+--------------+

关键说明 #

• 为什么不能直接对原表用窗口函数？因为原表同一日期的多条记录，直接排序后窗口会包含同一日期的其他记录，导致同一日期内的Y被误判为“之前的记录”，违反需求中“同一日期的多条记录均视为该日期前无Y”的规则。
• 聚合步骤确保了每个用户的每个日期只有一条判断结果，再通过窗口函数计算历史情况，最后关联回原表保证所有记录都能获取正确的标记值。

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