Spark中用Filter替代INNER JOIN的实现及底层优化探究

一、替换INNER JOIN为Filter+Explode的正确Scala写法 #

假设原JOIN逻辑是客户表中每条记录与日期表中满足start_date <= date <= end_date的日期行关联，当t_dates仅包含1-3天数据时，可以通过「收集日期到本地→过滤有效客户→展开日期」的方式替代JOIN，代码如下：

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, explode, lit}
import java.sql.Date

val spark = SparkSession.builder().getOrCreate()

// 1. 读取日期表并收集日期到Driver端（数据量极小，无内存压力）
val targetDates: Set[Date] = spark.table("t_dates")
  .select("date")
  .as[Date]
  .collect()
  .toSet

// 2. 计算目标日期的边界，先过滤掉完全不匹配的客户（减少后续计算量）
val minDate = targetDates.min
val maxDate = targetDates.max

val filteredCustomers = spark.table("reporting.customers")
  .filter(col("start_date") <= maxDate && col("end_date") >= minDate)

// 3. 为每个有效客户生成对应目标日期行，并过滤出真正符合日期范围的记录
val result = filteredCustomers
  .withColumn("date", explode(array(targetDates.map(lit(_)): _*)))
  .filter(col("start_date") <= col("date") && col("end_date") >= col("date"))

关键说明： #

• 先收集日期到本地集合：因为只有1-3天，完全不会占用Driver端过多内存。
• 前置过滤：通过日期边界快速排除与目标日期无交集的客户，避免后续生成无效中间数据。
• explode+二次过滤：模拟原JOIN的关联逻辑，确保最终结果与原INNER JOIN完全一致。

二、替换后的优化效果与OOM解决分析 #

1. 为什么能优化Spark执行？ #

原INNER JOIN如果未触发广播哈希连接（Broadcast Hash Join），会触发Shuffle操作：Spark需要对客户表和日期表进行分区、洗牌，过程中会产生大量磁盘IO和内存开销。而替换后的方案：

• 无Shuffle操作：所有计算都在客户表的原有分区内完成，避免了Shuffle阶段的内存占用。
• 数据量前置裁剪：通过日期边界过滤，直接减少了后续需要处理的客户数据量，降低内存负载。

2. 能否解决OOM问题？ #

大概率可以解决，原因如下：

• 原OOM通常发生在Shuffle阶段（如Shuffle Write时内存不足、或者Join时哈希表占用内存过大），替换后完全规避了Shuffle。
• 即使客户表数据量庞大，由于仅保留有效客户，且每个客户最多生成3行数据，中间数据量可控，不会出现内存溢出。

注意事项 #

• 如果客户表按start_date或end_date分区，可以在读取时加上分区裁剪（如spark.table("reporting.customers").where(col("start_date") <= maxDate)），进一步减少读取的数据量。
• 确保targetDates不为空，否则explode会生成空行，可提前判断并处理空值场景。

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