PySpark：如何更高效筛选DataFrame指定列最小值对应的行？

你的原方法虽然能实现需求，但通过collect()把最小值拉到Driver端，在大数据场景下会带来不必要的数据传输开销，推荐以下几种分布式执行的实现方式：

方法1：窗口函数（全局窗口计算） #

全程在集群端完成计算，无需将数据拉取到Driver，适合大规模数据集：

from pyspark.sql import Window
from pyspark.sql.functions import min, col

# 定义全局窗口（不做分区，计算整个DataFrame的最小值）
global_window = Window.partitionBy()
# 添加临时列存储全局最小值，筛选匹配行后删除临时列
df_min = df1.withColumn("global_min", min("col1").over(global_window)) \
            .filter(col("col1") == col("global_min")) \
            .drop("global_min")

方法2：子查询关联 #

直接在筛选逻辑中嵌套子查询获取最小值，同样是分布式执行：

from pyspark.sql.functions import min, col

# 直接在filter中使用子查询
df_min = df1.filter(col("col1") == df1.select(min("col1")).first()[0])

# 或者用join的方式，逻辑更清晰
min_subquery = df1.select(min("col1").alias("min_val"))
df_min = df1.join(min_subquery, df1["col1"] == min_subquery["min_val"], "inner") \
            .drop("min_val")

方法3：广播变量优化（适合重复使用最小值的场景） #

如果需要多次使用这个最小值进行筛选，可以将其广播到所有Executor，减少重复传输：

from pyspark.sql.functions import min, col
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

min_value = df1.select(min("col1")).collect()[0][0]
# 广播最小值到集群所有节点
broadcast_min = spark.sparkContext.broadcast(min_value)

df_min = df1.filter(col("col1") == broadcast_min.value)

各方法对比 #

• 窗口函数和子查询：无需将数据拉到Driver，完全分布式执行，是大数据场景下的首选方案。
• 广播变量：适合需要重复使用最小值的场景，相比原方法减少了值的传输次数，但仍需一次collect()。

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