Spark中如何将一个DataFrame的聚合结果用于过滤另一个DataFrame？

这里提供三种符合Spark分布式特性的标准实现方式，适配不同场景：

方式一：广播聚合结果（推荐） #

当聚合结果仅为单条记录时，将其广播到所有Executor节点，避免重复计算，同时贴合Spark分布式计算的设计思路。

import org.apache.spark.sql.functions._

// 对df1执行过滤与聚合，给聚合列命名方便后续引用
val aggDf1 = df1
  .filter(/* 你的过滤逻辑 */)
  .agg(min("A").alias("min_A"), max("A").alias("max_A"))

// 广播聚合结果DataFrame
val broadcastAgg = broadcast(aggDf1)

// 交叉连接后过滤，最后移除不需要的聚合列
val filteredDf2 = df2
  .crossJoin(broadcastAgg)
  .filter(col("myColumn") > col("min_A") && col("myColumn") < col("max_A"))
  .drop("min_A", "max_A")

方式二：提取聚合结果到Driver端（适合小数据量场景） #

如果能确定聚合结果仅为一行且数据量极小，可以将结果拉取到Driver端，再作为常量过滤df2。需注意处理聚合结果为空的情况：

import org.apache.spark.sql.functions._

val aggDf1 = df1
  .filter(/* 你的过滤逻辑 */)
  .agg(min("A"), max("A"))

// 安全提取聚合值，处理空结果的边界情况
val (minA, maxA) = aggDf1.headOption match {
  case Some(row) => (row.getAs[Int](0), row.getAs[Int](1))
  case None => (Int.MinValue, Int.MaxValue) // 可根据业务需求调整空值逻辑
}

// 用常量值过滤df2
val filteredDf2 = df2
  .filter(col("myColumn") > minA && col("myColumn") < maxA)

方式三：使用子查询（代码简洁） #

Spark 2.0+支持在过滤条件中直接嵌入子查询，代码最简洁，但需注意：若过滤逻辑复杂，会重复执行两次df1的过滤与聚合操作，性能略逊于方式一。

import org.apache.spark.sql.functions._

val filteredDf2 = df2.filter(
  col("myColumn") > (select(min("A")) from df1.filter(/* 你的过滤逻辑 */)) &&
  col("myColumn") < (select(max("A")) from df1.filter(/* 你的过滤逻辑 */))
)

场景选择建议 #

• 优先选方式一：适配绝大多数分布式场景，性能最优，避免Driver与Executor间的传输风险。
• 方式二：仅在聚合结果绝对为单条极小数据时使用，代码简洁但需处理空值。
• 方式三：适合快速编写原型代码，不推荐在性能敏感的生产环境使用。

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