Pandas及PySpark中分组获取速度最值对应车辆与驾驶员信息的高效方案

Pandas 解决方案 #

避免使用merge，窗口函数是更高效的选择，能直接在原数据集上标记分组内的极值排名，再筛选目标记录：

方法1：基于排名筛选完整记录 #

通过rank()函数给每个分组内的记录按速度排名，筛选排名为1的记录即可保留所有并列的最快/最慢数据：

import pandas as pd

# 假设原始DataFrame为df，包含列：Race_Event_ID、Car_ID、Driver_ID、AvgSpeed
# 给每个赛事分组内的速度降序排名（最快的排第1）
df['rank_max'] = df.groupby('Race_Event_ID')['AvgSpeed'].rank(method='dense', ascending=False)
# 给每个赛事分组内的速度升序排名（最慢的排第1）
df['rank_min'] = df.groupby('Race_Event_ID')['AvgSpeed'].rank(method='dense', ascending=True)

# 提取所有最快的记录
fastest_records = df[df['rank_max'] == 1][['Race_Event_ID', 'Car_ID', 'Driver_ID', 'AvgSpeed']]
# 提取所有最慢的记录
slowest_records = df[df['rank_min'] == 1][['Race_Event_ID', 'Car_ID', 'Driver_ID', 'AvgSpeed']]

• method='dense'确保并列的记录获得相同排名，不会跳过后续名次；如果只需取每组第一个极值记录，可改用method='first'。

方法2：分组聚合直接生成结果 #

如果需要将最快/最慢信息合并到同一条分组记录中，可直接在groupby.agg里通过lambda或条件筛选实现：

result = df.groupby('Race_Event_ID').agg(
    max_speed=('AvgSpeed', 'max'),
    fastest_cars=('Car_ID', lambda x: list(x[df.loc[x.index, 'AvgSpeed'] == x.max()])),
    fastest_drivers=('Driver_ID', lambda x: list(x[df.loc[x.index, 'AvgSpeed'] == x.max()])),
    min_speed=('AvgSpeed', 'min'),
    slowest_cars=('Car_ID', lambda x: list(x[df.loc[x.index, 'AvgSpeed'] == x.min()])),
    slowest_drivers=('Driver_ID', lambda x: list(x[df.loc[x.index, 'AvgSpeed'] == x.min()]))
).reset_index()

这种方式无需额外生成排名列，直接一次聚合完成，内存效率更高。

PySpark 解决方案 #

PySpark同样依赖窗口函数实现高效分组极值提取，核心逻辑与Pandas一致：

方法1：基于排名筛选完整记录 #

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

spark = SparkSession.builder.appName("RaceDataProcessing").getOrCreate()

# 定义窗口：按赛事分组，分别按速度降序/升序排序
window_max = Window.partitionBy("Race_Event_ID").orderBy(F.desc("AvgSpeed"))
window_min = Window.partitionBy("Race_Event_ID").orderBy(F.asc("AvgSpeed"))

# 添加排名列
spark_df = spark_df.withColumn("rank_max", F.dense_rank().over(window_max))
spark_df = spark_df.withColumn("rank_min", F.dense_rank().over(window_min))

# 筛选最快和最慢的记录
fastest_records = spark_df.filter(F.col("rank_max") == 1).select(
    "Race_Event_ID", "Car_ID", "Driver_ID", "AvgSpeed"
)
slowest_records = spark_df.filter(F.col("rank_min") == 1).select(
    "Race_Event_ID", "Car_ID", "Driver_ID", "AvgSpeed"
)

方法2：分组聚合合并结果 #

如果需要将极值对应的车辆、司机信息聚合到同一条记录，可结合collect_list和when条件实现：

result_spark = spark_df.groupBy("Race_Event_ID").agg(
    F.max("AvgSpeed").alias("max_speed"),
    F.collect_list(F.when(F.col("AvgSpeed") == F.max("AvgSpeed"), F.col("Car_ID"))).alias("fastest_cars"),
    F.collect_list(F.when(F.col("AvgSpeed") == F.max("AvgSpeed"), F.col("Driver_ID"))).alias("fastest_drivers"),
    F.min("AvgSpeed").alias("min_speed"),
    F.collect_list(F.when(F.col("AvgSpeed") == F.min("AvgSpeed"), F.col("Car_ID"))).alias("slowest_cars"),
    F.collect_list(F.when(F.col("AvgSpeed") == F.min("AvgSpeed"), F.col("Driver_ID"))).alias("slowest_drivers")
)

# 移除聚合结果中的None值（不满足条件的记录会返回None）
result_spark = result_spark.withColumn("fastest_cars", F.array_remove(F.col("fastest_cars"), None)) \
                          .withColumn("fastest_drivers", F.array_remove(F.col("fastest_drivers"), None)) \
                          .withColumn("slowest_cars", F.array_remove(F.col("slowest_cars"), None)) \
                          .withColumn("slowest_drivers", F.array_remove(F.col("slowest_drivers"), None))

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