04 · Join 策略与数据倾斜

Join 让匹配的记录相遇。如何移动数据，取决于两侧规模、键分布、输出基数和内存容量。只知道算法名字，无法诊断问题。

学习目标与前置知识

需要理解按键分区、Shuffle 与局部聚合，并知道内连接中每个匹配对贡献一条输出。本章比较广播与排序合并计划，检查热点分区，并评价加盐的代价。

决定让哪一侧移动

广播哈希 Join 从足够小的一侧建立查找结构，让处理大表的位置都能使用它。它用复制和构建内存换取大表免于按键重分布。磁盘上的压缩文件大小不等于内存哈希表大小，小表必须满足实际资源和连接类型的约束。

Shuffle Sort-Merge Join 将相同键重分布到兼容分区，并按需排序，能够处理更大的关系，但要支付移动与排序成本。哈希能分散不同键，却不能仅靠增加分区把一个巨大的相同键组拆开。Spark 的策略提示与 AQE 行为参见官方文档。Spark SQL 性能指南

原创概念图 · 不是运行轨迹。

受控对照

事实表有 240,000 行，其中 80% 使用键 0。维表有 1,000 行，每个键恰好一行。输出应为 240,000 行，权重总和 384,001。实验强制排序合并、显式广播维表，并尝试加盐排序合并。关闭 AQE 与自动广播，便于解释基线计划。

shuffle = fact.hint('merge').join(dim.hint('merge'), 'k')
broadcast = fact.join(F.broadcast(dim), 'k')
# 每条事实分配一个盐值，维表复制到八个盐桶。
left = fact.withColumn('salt', (F.col('i') % 8).cast('int'))
right = dim.crossJoin(spark.range(8).select(F.col('id').cast('int').alias('salt')))
salted = left.hint('merge').join(right.hint('merge'), ['k', 'salt'])

每条事实只匹配一次，是因为它的盐值恰好对应一份维表记录。随意复制两边会放大匹配数。本教学实现给所有键加盐，因此付出的复制成本高于只处理已知热点的策略。

先读证据，再选择

python scripts/run_all.py --only join

在物理计划中找 SortMergeJoin、BroadcastHashJoin 和 Exchange。实验直接按分区消费 Join 输出，避免后续聚合掩盖倾斜。未加盐时最大输出分区有 197,242 行；广播保留八个各 30,000 行的事实分区；加盐后最大分区降到 78,149 行，但哈希碰撞仍造成不均衡。本次加盐移动更多字节，中位耗时也没有改善。分布更均衡，不自动等于更快。

维表键唯一，因此行数与权重和是有用的检查，但校验和不是任意 Join 重写的普遍证明。自己的小样本应比较完整排序结果，覆盖重复键和空值。外连接还需要重新检查未匹配记录，不能直接照搬加盐。

动手与自测

固定总行数与维表，只改变热点比例；分别预测任务不均衡、输出数量和广播可行性。可选扩展中打开 AQE，保存初始与最终执行计划，检查阈值是否真正触发改写。本版讲解 AQE，但没有把该扩展描述成已实测结果。

过关任务

写出一份策略选择，包含关系基数、键唯一性、热点频率、内存预算与实际计划节点，并给出一个会推翻选择的条件。下一章考虑任务需要重跑的情况。