Spark 2.2.0关联DataFrame时触发8GB广播限制SparkException求助

嘿，这个问题我之前处理大数据join场景时也碰到过，咱们一步步拆解解决它～

问题根源 #

这个异常的核心逻辑很清晰：Spark的**广播哈希Join（Broadcast Hash Join）**有默认大小限制——在Spark 2.2.0里，spark.sql.autoBroadcastJoinThreshold参数默认值是8589934592字节（也就是8GB）。当Spark自动选择广播策略处理你的join操作时，发现要广播的DataFrame大小超过了这个阈值，就会抛出这个异常，进而导致整个作业终止（从你提供的堆栈里FileFormatWriter: Aborting job null也能印证这一点）。另外那条YarnAllocator: Driver requested a total number of 0 executor(s)的日志，是因为作业在广播阶段就失败了，还没来得及申请executor资源。

具体解决方案 #

1. 调整广播阈值（适合集群资源充足的情况） #

如果你的集群executor有足够内存容纳更大的广播表，可以直接调大这个阈值参数（注意参数值单位是字节）：

• 在代码中设置：

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10737418240") // 10GB

• 或者提交作业时通过命令行参数设置：

spark-submit --conf spark.sql.autoBroadcastJoinThreshold=10737418240 ...

⚠️ 注意：不要盲目调得过大！过大的广播表会占用每个executor的内存，可能引发OOM（内存溢出）问题，一定要根据集群实际资源情况来设置。

2. 强制使用非广播的Join策略 #

如果这个DataFrame确实太大不适合广播，直接告诉Spark不要用广播策略，改用适合大表的Join方式：

• 使用shuffle_hash hint（适合一张表略大、另一张表很大的场景）：

df1.join(df2.hint("shuffle_hash"), "join_key")

• 使用merge hint（适合两张都是大表的场景，需要表在join键上有序）：

df1.join(df2.hint("merge"), "join_key")

也可以直接禁用自动广播功能，设置spark.sql.autoBroadcastJoinThreshold=-1，这样Spark会根据表大小自动选择合适的Join策略，不会尝试自动广播任何表。

3. 先清洗数据缩小表体积 #

有时候表大小超标是因为存在重复数据、冗余列或无效数据，先做一步数据清洗再进行Join，可能就能把表压缩到广播阈值以内：

// 去重+只保留需要的列
val optimizedDf2 = df2.dropDuplicates("join_key")
                      .select("join_key", "required_col1", "required_col2")
df1.join(optimizedDf2, "join_key")

额外建议 #

Spark 2.2.0是比较老旧的版本（发布于2017年），如果条件允许，建议升级到Spark 3.x版本。新版本对Join策略的优化更智能，支持更多Join类型（比如Shuffle Replicate Join），阈值配置也更灵活，能避免很多老版本的坑。

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