嘿，你的问题我太熟悉了——小表左连超大表，结果大表全量扫描+shuffle直接把任务拖垮。结合你给的df2.explain输出和当前的Spark配置，我给你整理了几个优先级从高到低的优化步骤，照着做应该能把耗时从几小时降到几分钟：

1. 强制使用广播连接，彻底避免大表Shuffle #

df1只有500行，完全符合Spark广播连接的场景。虽然Spark默认会自动广播小表，但有时候自动判断可能失效，咱们手动指定更稳妥：

import org.apache.spark.sql.functions.broadcast
val result = broadcast(df1).join(df2, Seq("product_PK", "rec_product_PK"), "left")

广播后，Spark会把df1的全量数据分发到所有Executor节点，不需要对df2做任何Shuffle操作，直接在每个节点上做本地匹配，这能砍掉90%以上的无效数据传输。

2. 先过滤大表，只保留需要的极小部分数据 #

你的预期结果只有500行，说明df2里只有和df1匹配的少数行有用。咱们先从df1提取需要匹配的键对，再用它过滤df2，直接把6亿行的处理量压缩到几百行：

// 先从df1拿到所有需要匹配的(product_PK, rec_product_PK)组合
val targetKeys = df1.select("product_PK", "rec_product_PK").distinct()
// 过滤df2只保留匹配的行，再和df1做左连
val filteredDf2 = df2.join(broadcast(targetKeys), Seq("product_PK", "rec_product_PK"))
val result = df1.join(filteredDf2, Seq("product_PK", "rec_product_PK"), "left")

如果df2用的是Parquet这类支持谓词下推的存储格式，这一步还能让Spark直接跳过不相关的文件块，IO量能砍到原来的千分之一甚至更低。

3. 干掉df2不必要的重分区操作 #

从df2.explain的输出能看到，df2被强制重分区到了5000个分区：

Repartition 5000, true

这会导致6亿行数据全量Shuffle，纯纯的性能杀手！如果这个重分区不是业务必须的（比如之前代码里手动加的repartition(5000)），直接删掉就行；如果是读取数据时的默认配置，可以调整spark.sql.shuffle.partitions，但更关键的是先过滤再处理，根本不需要这么多分区。

4. 调整Spark资源和参数适配大表 #

当前的配置（2个Executor，每个4核10G内存）有点保守，建议根据集群资源调整：

• 增加num-executors：比如提到8-16个（看集群剩余资源），多节点并行处理能大幅提速
• 调整spark.sql.shuffle.partitions：默认是200，建议设置为Executor核数 * Executor数量的2-3倍（比如4核*8个Executor=32，设置为64-96），避免Shuffle时分区过小或过大
• 确保spark.sql.parquet.filterPushdown设置为true（默认开启），让过滤条件下推到存储层面

5. 后续优化：给df2合理分区 #

如果df2后续还要频繁做这类关联查询，建议按product_PK做分区存储。这样下次查询时，Spark可以直接只扫描匹配的分区，不用全表扫，性能会再上一个台阶。

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