Spark中如何无大量Shuffle连接同分区列的两个DataFrame？

针对两个按col1分区的DataFrame连接耗时过长的问题，核心原因是连接键为(col1, col2)，但现有分区仅基于col1，导致同一col1分区内的col2数据无法直接匹配，触发大量Shuffle。以下是几个可行的优化方案：

1. 重新按连接键分区 #

将两个DataFrame都按完整连接键(col1, col2)重新分区，确保相同连接键的数据落在同一个分区内，连接时无需跨分区Shuffle，直接在本地完成匹配。

选择分区数时，建议参考原DF的分区数（比如取两者的最大值120000），避免分区过小导致任务过多或分区过大导致数据倾斜：

// Scala 示例
val df1Opt = df1.repartition(120000, $"col1", $"col2")
val df2Opt = df2.repartition(120000, $"col1", $"col2")
val joinedDF = df1Opt.join(df2Opt, Seq("col1", "col2"), "inner")

# Python 示例
df1_opt = df1.repartition(120000, "col1", "col2")
df2_opt = df2.repartition(120000, "col1", "col2")
joined_df = df1_opt.join(df2_opt, on=["col1", "col2"], how="inner")

2. 广播小表（Broadcast Hash Join） #

如果其中一个DataFrame的数据量较小（比如df2分区数80000但单分区数据量不大，总数据量适合广播），可以将其广播到所有Executor节点，大表无需Shuffle，直接在本地与广播的小表进行连接。

Spark默认会自动判断小表是否适合广播，但也可以手动指定：

// Scala 示例
import org.apache.spark.sql.functions.broadcast
val joinedDF = df1.join(broadcast(df2), Seq("col1", "col2"), "inner")

# Python 示例
from pyspark.sql.functions import broadcast
joined_df = df1.join(broadcast(df2), on=["col1", "col2"], how="inner")

3. 调整分区粒度并匹配分区数 #

当前两个DF的分区数差异较大（120000 vs 80000），可能导致部分Executor负载不均。可以将两者的分区数调整为相同的合理值（比如取最小公倍数240000，或根据集群资源调整），同时按连接键分区，进一步优化连接效率。

4. 预处理过滤减少数据量 #

在连接前先对两个DF进行过滤，移除无效数据（如col1/col2为空的记录）或仅保留业务需要的数据集范围，减少后续连接的数据处理量，从根源降低耗时。

示例：

val df1Filtered = df1.filter($"col1".isNotNull && $"col2".isNotNull)
val df2Filtered = df2.filter($"col1".isNotNull && $"col2".isNotNull)

5. 使用分桶表（Bucketed Table） #

如果这两个DataFrame是从Hive表读取的，可以将源表按(col1, col2)分桶存储。Spark读取分桶表时会自动利用分桶信息，避免连接时的Shuffle操作，直接进行分桶内的匹配。

创建分桶表示例：

CREATE TABLE bucketed_df1 (col1 string, col2 string, col3 string)
CLUSTERED BY (col1, col2) INTO 120000 BUCKETS;

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