PySpark按月份分组求销量最大值并展示对应ID的实现方法

直接用groupby+agg只能得到月份和对应销量最大值，没法直接关联到对应的ID，因为分组聚合不会保留非聚合列的匹配关系。下面提供两种可行的解决方案：

方法一：窗口函数法（推荐） #

利用窗口函数对每个月份分组后按销量降序排序，取每组的第一条记录，就能同时拿到ID和最大销量：

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

# 定义窗口规则：按month分组，按units_sold降序排列
window_spec = Window.partitionBy("month").orderBy(F.desc("units_sold"))

# 添加行号列，筛选出每组的第一条记录，整理列名
result_df = df.withColumn("row_num", F.row_number().over(window_spec)) \
              .filter(F.col("row_num") == 1) \
              .select("month", "id", F.col("units_sold").alias("max(units_sold)"))

# 查看结果
result_df.show()

如果同一个月份存在多个ID的销量同为最大值，row_number()只会保留其中一条。若要保留所有最大值记录，可将row_number()替换为rank()或dense_rank()。

方法二：聚合后关联法 #

先单独计算每个月份的最大销量，再通过关联原表匹配到对应的ID：

import pyspark.sql.functions as F

# 第一步：计算每个月份的最大销量
max_sold_df = df.groupBy("month").agg(F.max("units_sold").alias("max(units_sold)"))

# 第二步：关联原表，匹配月份和销量都对应的记录
result_df = df.join(max_sold_df, 
                    (df["month"] == max_sold_df["month"]) & (df["units_sold"] == max_sold_df["max(units_sold)"]),
                    "inner") \
              .select(max_sold_df["month"], "id", "max(units_sold)")

# 查看结果
result_df.show()

这种方法会保留所有销量等于当月最大值的记录，适合需要展示全部极值对应ID的场景。

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