Spark SQL 1.6.2多表Join最后阶段失败问题排查求助

嘿，针对你遇到的这个问题——用Spark SQL 1.6.2写的作业，23个规模相近的表（每个约550万行）用几乎唯一的字符串键连接，最后阶段执行失败，我结合这个版本的特性和你的场景，整理了几个实用的解决思路，应该能帮你搞定：

一、先优化Join的执行顺序，别让优化器瞎选 #

Spark 1.6.2的Catalyst优化器在处理多表Join时，自动选择的执行顺序不一定最优，尤其是面对23个表这么大的规模，很容易导致中间结果集爆炸，最后扛不住崩溃。

• 手动控制Join顺序：别一次性把所有表丢进一个Join里，试着分步合并——先把两个表Join得到中间结果，再和第三个表Join，以此类推。优先Join那些合并后结果集增长最小的表，一步步累加，避免中间数据量突然暴涨。
• 试试Broadcast Join：虽然你说每张表规模相近，但可以检查下有没有相对偏小的表（比如某些表实际数据量比其他表少），用/*+ BROADCAST(tableName) */提示强制广播这个表。Spark 1.6默认广播阈值是10MB，要是需要可以调大spark.sql.autoBroadcastJoinThreshold，不过别调得太夸张，不然Driver内存会吃不消。

二、把字符串连接键换成数值类型，减少shuffle开销 #

字符串类型的键可比数值类型费资源多了——序列化占空间大，哈希计算也慢，多表Join的shuffle阶段这个开销会被放大好几倍。既然你的键几乎唯一，重复值才7个以内，完全可以转成数值ID来做Join：

-- 第一步：从所有表提取唯一键，生成字符串到ID的映射表
CREATE TEMPORARY VIEW key_mapping AS
SELECT DISTINCT join_key, MONOTONICALLY_INCREASING_ID() AS key_id
FROM (
    SELECT join_key FROM table1
    UNION ALL SELECT join_key FROM table2
    -- ... 把剩下21个表都加进来
) all_keys;

-- 第二步：给每个表的字符串键替换成数值ID
CREATE TEMPORARY VIEW table1_encoded AS
SELECT t.*, m.key_id
FROM table1 t
JOIN key_mapping m ON t.join_key = m.join_key;

-- 其他22个表照猫画虎，之后用key_id来做Join就行

换成数值ID后，shuffle时的数据量会大幅减少，Executor的内存压力也会小很多，大概率能解决最后阶段的崩溃问题。

三、调大资源和shuffle参数，给作业松绑 #

最后阶段失败大概率是shuffle时内存不够或者资源瓶颈，针对Spark 1.6.2可以调整这些参数：

• 加Executor内存和核心数：如果集群有多余资源，把spark.executor.memory从默认的1G调到4G甚至8G，同时增加spark.executor.cores，让每个Executor能扛更多数据。
• 给shuffle多分点内存：Spark 1.6默认shuffle内存只占Executor内存的20%，可以把spark.shuffle.memoryFraction调到30%或40%，减少磁盘溢写的概率，毕竟溢写太费时间还容易出错。
• 确保shuffle压缩开启：检查spark.shuffle.compress和spark.shuffle.spill.compress是不是设为true（默认是开启的），压缩shuffle数据能减少磁盘IO和网络传输量，速度会快很多。
• 调整shuffle分区数：默认是200个分区，对于你的场景可能不够或者太多。可以根据Executor核心数来调，比如每个Executor有4核，集群有10个Executor，就设成80-120个分区（核心数的2-3倍），让每个分区的数据量更均衡，不会出现某个Task累死的情况。

四、排查下有没有隐藏的数据倾斜（虽然概率低，但还是要确认） #

你说连接键重复值不超过7个，但还是查一下放心——对每个表跑个统计：

SELECT join_key, COUNT(*) FROM tableX GROUP BY join_key ORDER BY COUNT(*) DESC LIMIT 10;

看看有没有某个键的行数远超其他，要是有的话，就得单独处理这个键（比如拆分Task），不过根据你的描述这种情况应该很少见。

五、把大Join拆成多个小Join #

23个表一起Join太猛了，拆成阶段来做——比如先每5个表一组Join，得到中间结果后再和其他组合并，这样每次处理的数据量小，内存压力也小，不容易挂掉。

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