在Spark中,join操作用于合并两个数据集(如dataFrame和Dataset),其原理依赖于分布式计算的特性。Spark的join主要通过不同的连接策略来实现,选择哪种策略取决于多种因素,包括参数连接的数据集大小,是否进行了广播,是否有共同的分区分区键等。
1. Shuffle Hash Join
这是最常用的连接方式之一。当两个数据集都非常大且无法被广播时,Spark会使用shuffle hash join。这种连接方式的工作流程如下:
- 数据按照连接键进行哈希分区。
- 分区后的数据会在集群中重新分布。(shuffle)
- 在每个工作节点上,对本地分区的数据执行连接操作。
这种方式的优点是通用性强,可以处理任意大小的数据集;缺点是需要进行全量的shuffle操作,这可能会带来较大的性能开销。
2. Broadcast Hash Join
当其中一个数据集相对较小(通常小于spark.sql.autoBroadcastJoinThreshold设置的阈值,默认为10M),spark可以将这个小的数据集广播到所有的工作节点上,然后在每个节点上与另一个大数据集进行连接。这种方式避免了shuffle操作,因此通常比shuffle hash join更快,但前提是广播的数据集必须足够小,以适应所有节点的内存。
3. Sort Merge Join
如果两个数据集都已经按照连接键排序,并且这些数据集很大,不适合广播,那么spark将采用sort Merge join。在这种情况下,spark首先会对两个数据集按照连接键排序,然后进行合并操作。此过程同样涉及到shuffle操作,因为数据可能需要重新分区以确保相同的键值的数据位于同一分区。sort Merge Join对于有序的数据集来说非常高效,但它也要求额外的排序步骤,这可能会增加一定的计算成本。
4. 笛卡尔积
当没有指定连接条件的时候,spark会执行笛卡尔积,即返回两个数据集所有行的组合。这种连接方式很少用,因为它会导致输出结果的数量急剧膨胀,通常是第一个数据集的行数乘上第二个数据集的行数。
5. 自适应查询执行和Join调优
从spark3.x开始引入的AQE功能可以在运行时动态优化join操作,例如,AQE可以根据实际的数据量决定是否应该使用广播连接,或者自动调整shuffle partitions的数量以优化性能。