PySpark实现按商品-日期分组插补销量负数值的最优方案咨询

方案1：已有预计算众数映射表的实现 #

如果你已经提前统计好了每个商品-日期组合对应的最高频销量值，直接关联两张表做条件替换即可，步骤如下：

• 假设你的原始销售表为sales_raw，包含字段：商品id、日期、销量
• 假设预计算的众数映射表为sales_mode，包含字段：商品id、日期、高频销量值

from pyspark.sql import functions as F

# 关联原始表和众数映射表
sales_joined = sales_raw.join(
    sales_mode,
    on=["商品id", "日期"],
    how="left"
)

# 负数值替换为对应众数，非负值保留原值
sales_filled = sales_joined.withColumn(
    "销量",
    F.when(F.col("销量") < 0, F.col("高频销量值")).otherwise(F.col("销量"))
).drop("高频销量值")

方案2：无预计算众数，单表一步完成插补 #

如果还没单独统计众数，直接用窗口函数在原始表操作即可，避免额外的shuffle开销，性能更优：

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

# 定义分组窗口：按商品id和日期分组
win = Window.partitionBy("商品id", "日期")

# 统计每个销量值在对应组的出现次数
sales_with_cnt = sales_raw.withColumn(
    "sale_cnt",
    F.count("销量").over(win.partitionBy("商品id", "日期", "销量"))
)

# 取每个组出现次数最高的销量作为众数
sales_with_mode = sales_with_cnt.withColumn(
    "sale_mode",
    F.max(F.struct(F.col("sale_cnt"), F.col("销量"))).over(win).getField("销量")
)

# 替换负数值，清理中间字段
sales_filled = sales_with_mode.withColumn(
    "销量",
    F.when(F.col("销量") < 0, F.col("sale_mode")).otherwise(F.col("销量"))
).drop("sale_cnt", "sale_mode")

边界处理说明 #

• 如果某个商品-日期组的所有销量都是负数，众数也会是负数，你可以额外加判断逻辑，将这类情况的负数值替换为0或其他自定义默认值
• 如果同组有多个销量出现次数相同（即多个众数），上述代码会自动取数值更大的那个众数，如果要取更小的，把代码中的max改成min即可

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