关于Spark报“无法广播大于8GB的表”错误的疑问澄清

先明确你的场景伪代码：

val riskDF = broadcast(someDF) // 期望riskDF作为广播表，大小小于1GB

val processDF1 = ProcessAndJoin(riskDF) // 读取其他数据源并关联riskDF
val processDF2 = ProcessAndJoin(riskDF) // 读取其他数据源并关联riskDF
val processDF3 = ProcessAndJoin(riskDF) // 读取其他数据源并关联riskDF

合并processDF1/2/3后写入存储桶，且已开启spark.sql.adaptive.enabled=true。

针对你的疑问逐一解答：

疑问1：riskDF被多次读取和广播是否正常？ #

你原本认为只会广播一次的逻辑是合理的，但开启AQE（自适应执行）后确实可能出现重复广播：

• AQE会对每个独立的执行分支（比如processDF1/2/3各自的执行链）做自适应优化，分支之间的广播操作可能被独立触发，而非共享同一个广播变量。
• 如果someDF没有被显式持久化（比如调用persist），每次引用riskDF都会重新计算其数据，进而触发重复广播。即使加了broadcast hint，AQE的分支独立优化逻辑也可能打破单次广播的预期。

疑问2：8GB限制是单次还是累计？ #

这个限制是针对单次广播的表大小，和整个流水线的累计广播量无关。
Spark源码中，BroadcastExchangeExec会检查待广播表的统计大小，一旦超过8GB（硬编码阈值）就会抛出该错误。这个检查只针对当前要广播的单个表，不会累加所有广播表的大小。你的场景中riskDF本身小于1GB，出现错误大概率是因为AQE重复广播时，某个分支的统计计算出现偏差（比如数据倾斜或统计信息不准确），导致Spark误判表大小超过8GB。

疑问3：该错误是否与spark.sql.autoBroadcastJoinThreshold无关？ #

你的理解完全正确。
spark.sql.autoBroadcastJoinThreshold是Spark自动判断是否广播小表的阈值，而broadcast hint是显式强制广播指定表，两者逻辑独立。即使你没修改这个参数的默认值，只要用了broadcast hint，Spark就会忽略自动广播阈值，强制尝试广播该表，所以这个错误和该参数无关。

关于关闭AQE的尝试 #

你尝试设置spark.sql.adaptive.enabled=false是合理的：关闭AQE后，Spark会生成统一的执行计划，更容易实现riskDF的单次广播复用，大概率能解决重复广播带来的误判问题。

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