Join是SQL语句中的常用操作,良好的表结构能够将数据分散在不同的表中,使其符合某种范式,减少表冗余、更新容错等。而建立表和表之间关系的最佳方式就是Join操作。
SparkSQL作为大数据领域的SQL实现,自然也对Join操作做了不少优化,今天主要看一下在SparkSQL中对于Join中常见的3种实现。
BroadcastJoin
大家知道,在数据库的常见模型中(比如星型模型或者雪花模型),表一般分为两种:事实表和维度表。维度表一般指固定的、变动较少的表,例如联系人、物品种类等,一般数据有限。而事实表一般记录流水,比如销售清单等,通常随着时间的增长不断膨胀。
因为Join操作是对两个表中key值相同的记录进行连接,在SparkSQL中,对两个表做Join最直接的方式是先根据key分区,再在每个分区中把key值相同的记录拿出来做连接操作。但这样就不可避免地涉及到shuffle,而shuffle在Spark中是比较耗时的操作,我们应该尽可能的设计Spark应用使其避免大量的shuffle。
当维度表和事实表进行Join操作时,为了避免shuffle,我们可以将大小有限的维度表的全部数据分发到每个节点上,供事实表使用。executor存储维度表的全部数据,一定程度上牺牲了空间,换取shuffle操作大量的耗时,这在SparkSQL中称作BroadcastJoin,如下图所示:
TableB是较小的表,黑色表示将其广播到每个executor节点上,TableA的每个partition会通过blockmanager取到TableA的数据。根据每条记录的JoinKey取到TableB中相对应的记录,根据JoinType进行操作。这个过程比较简单,不做赘述。
BroadcastJoin的条件有以下几个:
1.被广播的表需要小于spark.sql.autoBroadcastJoinThreshold所配置的值,默认是10M(或者加了broadcastjoin的hint)。
2.基表不能被广播,比如leftouterjoin时,只能广播右表。
看起来广播是一个比较理想的方案,但它有没有缺点呢?也很明显。这个方案只能用于广播较小的表,否则数据的冗余传输就远大于shuffle的开销;另外,广播时需要将被广播的表现collect到driver端,当频繁有广播出现时,对driver的内存也是一个考验。
ShuffleHashJoin
当一侧的表比较小时,我们选择将其广播出去以避免shuffle,提高性能。但因为被广播的表首先被collect到driver段,然后被冗余分发到每个executor上,所以当表比较大时,采用broadcastjoin会对driver端和executor端造成较大的压力。
但由于Spark是一个分布式的计算引擎,可以通过分区的形式将大批量的数据划分成n份较小的数据集进行并行计算。这种思想应用到Join上便是ShuffleHashJoin了。利用key相同必然分区相同的这个原理,SparkSQL将较大表的join分而治之,先将表划分成n个分区,再对两个表中相对应分区的数据分别进行HashJoin,这样即在一定程度上减少了driver广播一侧表的压力,也减少了executor端取整张被广播表的内存消耗。其原理如下图:
ShuffleHashJoin分为两步:
1.对两张表分别按照joinkeys进行重分区,即shuffle,目的是为了让有相同joinkeys值的记录分到对应的分区中。
2.对对应分区中的数据进行join,此处先将小表分区构造为一张hash表,然后根据大表分区中记录的joinkeys值拿出来进行匹配。
ShuffleHashJoin的条件有以下几个:
1.分区的平均大小不超过spark.sql.autoBroadcastJoinThreshold所配置的值,默认是10M
2.基表不能被广播,比如leftouterjoin时,只能广播右表
3.一侧的表要明显小于另外一侧,小的一侧将被广播(明显小于的定义为3倍小,此处为经验值)
我们可以看到,在一定大小的表中,SparkSQL从时空结合的角度来看,将两个表进行重新分区,并且对小表中的分区进行hash化,从而完成join。在保持一定复杂度的基础上,尽量减少driver和executor的内存压力,提升了计算时的稳定性。
SortMergeJoin
上面介绍的两种实现对于一定大小的表比较适用,但当两个表都非常大时,显然无论适用哪种都会对计算内存造成很大压力。这是因为join时两者采取的都是hashjoin,是将一侧的数据完全加载到内存中,使用hashcode取joinkeys值相等的记录进行连接。
当两个表都非常大时,SparkSQL采用了一种全新的方案来对表进行Join,即SortMergeJoin。这种实现方式不用将一侧数据全部加载后再进星hashjoin,但需要在join前将数据排序。
SparkSQL作为大数据领域的SQL实现,自然也对Join操作做了不少优化,今天主要看一下在SparkSQL中对于Join中常见的3种实现。
BroadcastJoin
大家知道,在数据库的常见模型中(比如星型模型或者雪花模型),表一般分为两种:事实表和维度表。维度表一般指固定的、变动较少的表,例如联系人、物品种类等,一般数据有限。而事实表一般记录流水,比如销售清单等,通常随着时间的增长不断膨胀。
因为Join操作是对两个表中key值相同的记录进行连接,在SparkSQL中,对两个表做Join最直接的方式是先根据key分区,再在每个分区中把key值相同的记录拿出来做连接操作。但这样就不可避免地涉及到shuffle,而shuffle在Spark中是比较耗时的操作,我们应该尽可能的设计Spark应用使其避免大量的shuffle。
当维度表和事实表进行Join操作时,为了避免shuffle,我们可以将大小有限的维度表的全部数据分发到每个节点上,供事实表使用。executor存储维度表的全部数据,一定程度上牺牲了空间,换取shuffle操作大量的耗时,这在SparkSQL中称作BroadcastJoin,如下图所示:
TableB是较小的表,黑色表示将其广播到每个executor节点上,TableA的每个partition会通过blockmanager取到TableA的数据。根据每条记录的JoinKey取到TableB中相对应的记录,根据JoinType进行操作。这个过程比较简单,不做赘述。
BroadcastJoin的条件有以下几个:
1.被广播的表需要小于spark.sql.autoBroadcastJoinThreshold所配置的值,默认是10M(或者加了broadcastjoin的hint)。
2.基表不能被广播,比如leftouterjoin时,只能广播右表。
看起来广播是一个比较理想的方案,但它有没有缺点呢?也很明显。这个方案只能用于广播较小的表,否则数据的冗余传输就远大于shuffle的开销;另外,广播时需要将被广播的表现collect到driver端,当频繁有广播出现时,对driver的内存也是一个考验。
ShuffleHashJoin
当一侧的表比较小时,我们选择将其广播出去以避免shuffle,提高性能。但因为被广播的表首先被collect到driver段,然后被冗余分发到每个executor上,所以当表比较大时,采用broadcastjoin会对driver端和executor端造成较大的压力。
但由于Spark是一个分布式的计算引擎,可以通过分区的形式将大批量的数据划分成n份较小的数据集进行并行计算。这种思想应用到Join上便是ShuffleHashJoin了。利用key相同必然分区相同的这个原理,SparkSQL将较大表的join分而治之,先将表划分成n个分区,再对两个表中相对应分区的数据分别进行HashJoin,这样即在一定程度上减少了driver广播一侧表的压力,也减少了executor端取整张被广播表的内存消耗。其原理如下图:
ShuffleHashJoin分为两步:
1.对两张表分别按照joinkeys进行重分区,即shuffle,目的是为了让有相同joinkeys值的记录分到对应的分区中。
2.对对应分区中的数据进行join,此处先将小表分区构造为一张hash表,然后根据大表分区中记录的joinkeys值拿出来进行匹配。
ShuffleHashJoin的条件有以下几个:
1.分区的平均大小不超过spark.sql.autoBroadcastJoinThreshold所配置的值,默认是10M
2.基表不能被广播,比如leftouterjoin时,只能广播右表
3.一侧的表要明显小于另外一侧,小的一侧将被广播(明显小于的定义为3倍小,此处为经验值)
我们可以看到,在一定大小的表中,SparkSQL从时空结合的角度来看,将两个表进行重新分区,并且对小表中的分区进行hash化,从而完成join。在保持一定复杂度的基础上,尽量减少driver和executor的内存压力,提升了计算时的稳定性。
SortMergeJoin
上面介绍的两种实现对于一定大小的表比较适用,但当两个表都非常大时,显然无论适用哪种都会对计算内存造成很大压力。这是因为join时两者采取的都是hashjoin,是将一侧的数据完全加载到内存中,使用hashcode取joinkeys值相等的记录进行连接。
当两个表都非常大时,SparkSQL采用了一种全新的方案来对表进行Join,即SortMergeJoin。这种实现方式不用将一侧数据全部加载后再进星hashjoin,但需要在join前将数据排序。