SparkSQL – 有必要坐下来聊聊Join

Join背景介绍

Join是数据库查询永远绕不开的话题,传统查询SQL技术总体可以分为简单操作(过滤操作-where、排序操作-limit等),聚合操作-groupBy等以及Join操作等。其中Join操作是其中最复杂、代价最大的操作类型,也是OLAP场景中使用相对较多的操作。因此很有必要聊聊这个话题。

另外,从业务层面来讲,用户在数仓建设的时候也会涉及Join使用的问题。通常情况下,数据仓库中的表一般会分为”低层次表”和“高层次表”。

所谓”低层次表”,就是数据源导入数仓之后直接生成的表,单表列值较少,一般可以明显归为维度表或者事实表,表和表之间大多存在外健依赖,所以查询起来会遇到大量Join运算,查询效率相对比较差。而“高层次表”是在”低层次表”的基础上加工转换而来,通常做法是使用SQL语句将需要Join的表预先进行合并形成“宽表”,在宽表上的查询因为不需要执行大量Join因而效率相对较高,很明显,宽表缺点是数据会有大量冗余,而且生成相对比较滞后,查询结果可能并不及时。

因此,为了获得实效性更高的查询结果,大多数场景还是需要进行复杂的Join操作。Join操作之所以复杂,不仅仅因为通常情况下其时间空间复杂度高,更重要的是它有很多算法,在不同场景下需要选择特定算法才能获得最好的优化效果。关系型数据库也有关于Join的各种用法,姜承尧大神之前由浅入深地介绍过MySQL Join的各种算法以及调优方案(关注公众号InsideMySQL并回复join可以查看相关文章)。本文接下来会介绍SparkSQL所支持的几种常见的Join算法以及其适用场景。

Join常见分类以及基本实现机制

当前SparkSQL支持三种Join算法-shuffle hash join、broadcast hash join以及sort merge join。其中前两者归根到底都属于hash join,只不过在hash join之前需要先shuffle还是先broadcast。其实,这些算法并不是什么新鲜玩意,都是数据库几十年前的老古董了(参考),只不过换上了分布式的皮而已。不过话说回来,SparkSQL/Hive…等等,所有这些大数据技术哪一样不是来自于传统数据库技术,什么语法解析AST、基于规则优化(CRO)、基于代价优化(CBO)、列存,都来自于传统数据库。就拿shuffle hash join和broadcast hash join来说,hash join算法就来自于传统数据库,而shuffle和broadcast是大数据的皮,两者一结合就成了大数据的算法了。因此可以这样说,大数据的根就是传统数据库,传统数据库人才可以很快的转型到大数据。好吧,这些都是闲篇。

继续来看技术,既然hash join是’内核’,那就刨出来看看,看完把’皮’再分析一下。

Hash Join

先来看看这样一条SQL语句:select * from order,item where item.id = order.i_id,很简单一个Join节点,参与join的两张表是item和order,join key分别是item.id以及order.i_id。现在假设这个Join采用的是hash join算法,整个过程会经历三步:

1. 确定Build Table以及Probe Table:这个概念比较重要,Build Table使用join key构建Hash Table,而Probe Table使用join key进行探测,探测成功就可以join在一起。通常情况下,小表会作为Build Table,大表作为Probe Table。此事例中item为Build Table,order为Probe Table。

2. 构建Hash Table:依次读取Build Table(item)的数据,对于每一行数据根据join key(item.id)进行hash,hash到对应的Bucket,生成hash table中的一条记录。数据缓存在内存中,如果内存放不下需要dump到外存。

3. 探测:再依次扫描Probe Table(order)的数据,使用相同的hash函数映射Hash Table中的记录,映射成功之后再检查join条件(item.id = order.i_id),如果匹配成功就可以将两者join在一起。



001

基本流程可以参考上图,这里有两个小问题需要关注:

1. hash join性能如何?很显然,hash join基本都只扫描两表一次,可以认为o(a+b),较之最极端的笛卡尔集运算a*b,不知甩了多少条街

2. 为什么Build Table选择小表?道理很简单,因为构建的Hash Table最好能全部加载在内存,效率最高;这也决定了hash join算法只适合至少一个小表的join场景,对于两个大表的join场景并不适用;

上文说过,hash join是传统数据库中的单机join算法,在分布式环境下需要经过一定的分布式改造,说到底就是尽可能利用分布式计算资源进行并行化计算,提高总体效率。hash join分布式改造一般有两种经典方案:

1. broadcast hash join:将其中一张小表广播分发到另一张大表所在的分区节点上,分别并发地与其上的分区记录进行hash join。broadcast适用于小表很小,可以直接广播的场景。

2. shuffler hash join:一旦小表数据量较大,此时就不再适合进行广播分发。这种情况下,可以根据join key相同必然分区相同的原理,将两张表分别按照join key进行重新组织分区,这样就可以将join分而治之,划分为很多小join,充分利用集群资源并行化。

Broadcast Hash Join

如下图所示,broadcast hash join可以分为两步:

1. broadcast阶段:将小表广播分发到大表所在的所有主机。广播算法可以有很多,最简单的是先发给driver,driver再统一分发给所有executor;要不就是基于bittorrete的p2p思路;

2. hash join阶段:在每个executor上执行单机版hash join,小表映射,大表试探;

002

SparkSQL规定broadcast hash join执行的基本条件为被广播小表必须小于参数spark.sql.autoBroadcastJoinThreshold,默认为10M。

Shuffle Hash Join

在大数据条件下如果一张表很小,执行join操作最优的选择无疑是broadcast hash join,效率最高。但是一旦小表数据量增大,广播所需内存、带宽等资源必然就会太大,broadcast hash join就不再是最优方案。此时可以按照join key进行分区,根据key相同必然分区相同的原理,就可以将大表join分而治之,划分为很多小表的join,充分利用集群资源并行化。如下图所示,shuffle hash join也可以分为两步:

1. shuffle阶段:分别将两个表按照join key进行分区,将相同join key的记录重分布到同一节点,两张表的数据会被重分布到集群中所有节点。这个过程称为shuffle

2. hash join阶段:每个分区节点上的数据单独执行单机hash join算法。



003

看到这里,可以初步总结出来如果两张小表join可以直接使用单机版hash join;如果一张大表join一张极小表,可以选择broadcast hash join算法;而如果是一张大表join一张小表,则可以选择shuffle hash join算法;那如果是两张大表进行join呢?

Sort-Merge Join

SparkSQL对两张大表join采用了全新的算法-sort-merge join,如下图所示,整个过程分为三个步骤:

004

1. shuffle阶段:将两张大表根据join key进行重新分区,两张表数据会分布到整个集群,以便分布式并行处理

2. sort阶段:对单个分区节点的两表数据,分别进行排序

3. merge阶段:对排好序的两张分区表数据执行join操作。join操作很简单,分别遍历两个有序序列,碰到相同join key就merge输出,否则取更小一边,见下图示意:

005

仔细分析的话会发现,sort-merge join的代价并不比shuffle hash join小,反而是多了很多。那为什么SparkSQL还会在两张大表的场景下选择使用sort-merge join算法呢?这和Spark的shuffle实现有关,目前spark的shuffle实现都适用sort-based shuffle算法,因此在经过shuffle之后partition数据都是按照key排序的。因此理论上可以认为数据经过shuffle之后是不需要sort的,可以直接merge。

经过上文的分析,可以明确每种Join算法都有自己的适用场景,数据仓库设计时最好避免大表与大表的join查询,SparkSQL也可以根据内存资源、带宽资源适量将参数spark.sql.autoBroadcastJoinThreshold调大,让更多join实际执行为broadcast hash join。

总结

Join操作是传统数据库中的一个高级特性,尤其对于当前MySQL数据库更是如此,原因很简单,MySQL对Join的支持目前还比较有限,只支持Nested-Loop Join算法,因此在OLAP场景下MySQL是很难吃的消的,不要去用MySQL去跑任何OLAP业务,结果真的很难看。不过好消息是MySQL在新版本要开始支持Hash Join了,这样也许在将来也可以用MySQL来处理一些小规模的OLAP业务。

和MySQL相比,PostgreSQL、SQLServer、Oracle等这些数据库对Join支持更加全面一些,都支持Hash Join算法。由PostgreSQL作为内核构建的分布式系统Greenplum更是在数据仓库中占有一席之地,这和PostgreSQL对Join算法的支持其实有很大关系。

总体而言,传统数据库单机模式做Join的场景毕竟有限,也建议尽量减少使用Join。然而大数据领域就完全不同,Join是标配,OLAP业务根本无法离开表与表之间的关联,对Join的支持成熟度一定程度上决定了系统的性能,夸张点说,’得Join者得天下’。本文只是试图带大家真正走进Join的世界,了解常用的几种Join算法以及各自的适用场景。后面两篇文章还会涉及Join的方方面面,敬请期待!

范欣欣

就职于网易杭州研究院后台技术中心数据库技术组,从事HBase开发、运维,对HBase相关技术有浓厚的兴趣。邮箱:libisthanks@gmail.com

在 “SparkSQL – 有必要坐下来聊聊Join” 上有 19 条评论

  1. 范神,Shuffle Hash Joinx下面的那张图,是Sort-Merge Join的图吧,Shuffle Hash Join的图是不是缺失了;
    cost(broadcast hash join) < cost(shuffle hash join)的原因是?

    1. 嗯嗯嗯 图有点错乱 修改过来了;现在集群的瓶颈一般都在内存和CPU,带宽相对来说比较充裕,broadcast代价一般会小于shuffle代价(消耗CPU+内存资源),只能说通常情况下一个表很小的情况下,braodcast代价会小于shuffle,当然,大表情况下肯定不成立。反复想了下,这句话不是很严谨,删了

  2. 范兄,你说的大表关联大表下 merge sort join 要比 Shuffle Hash Join 效率高这个不对吧:
    1:A hash join B COST=log(scan A)+log(hash A)+log(scan B)
    2:A merge sort join B COST=log(scan A)+log(sort A)+log(scan B)+log(sort B)+log(merge A+B)
    从算法分析看,A+B表的sort阶段要比A表hash build的消耗要高(不管A是否完全能放到内存)明显是1的代价小很多。

    1. 除非前提是A表和B表的数据默认已经排好序,这种情况2比1COST要小。所以要说快,是底层存储数据的顺序性导致这个算法快,离开底层数据排序方式说算法,有点强求:)。

      1. 原文中增加了:
        “仔细分析的话会发现,sort-merge join的代价并不比shuffle hash join小,反而是多了很多。那为什么SparkSQL还会在两张大表的场景下选择使用sort-merge join算法呢?这和Spark的shuffle实现有关,目前spark的shuffle实现都适用sort-based shuffle算法,因此在经过shuffle之后partition数据都是按照key排序的。因此理论上可以认为数据经过shuffle之后是不需要sort的,可以直接merge。”

    2. 真心谢谢一起交流!分析的绝对没毛病,正如你所言,正常情况下hash join确实会比merge sort join代价低得多。但spark这里会有一些特别,因为spark目前的shuffle算法都是sort-based shuffle,shuffle之后的数据都是经过排序的。这里我的表述有问题,需要做一些修改。
      参考:
      1. https://github.com/apache/spark/pull/5208
      2. https://github.com/apache/spark/pull/3438
      3. http://www.cnblogs.com/jcchoiling/p/6440102.html

  3. 博文思路清晰,通俗易懂,有个问题请教您,往往查询过程中,涉及到多个大表查询的场景下,这种情况下,I/O扫描的时间是不是小于中间大表之间的join的时间呢?还是说主要瓶颈还是在I/O扫描呢?

    1. spark中join如果是broadcast hash join的话性能还是不错的,可以pipeline执行。与之相对的shuffle hash join和sort merge join,后两者都有shuffle。shuffle是spark里面最耗时的操作,shuffle你就知道性能瓶颈在于什么地方了。所以说,join中包含但不止于IO扫描,还有网络传输、数据序列化、sort、hash等等过程,只能说瓶颈在于shuffle

      1. 可能之前没表达清楚,比如一个查询到几个大表,大表之间的join相比于最初scan table的话,哪个开销更大一些,之前说的扫描与shuffle无关,就是scan table。想问您的也是scan table vs join(sort merge join) 哪个代价更大更耗时。

        1. 那肯定还是join代价更大、消耗更多。join是一个复杂的过程,包含shuffle,一旦有shuffle,消耗肯定大的…

  4. 您好,在sparksql中两个表之间进行join操作时,可以指定join的方式吗?比如我的两个dataframe都比较小,想指定Broadcast join方式进行join。

  5. 你好,请问spark shuffle过程中数据是直接落磁盘吗?还是说shuffle数据先放在内存中,内存中如果放不下才落磁盘?

发表评论

电子邮件地址不会被公开。 必填项已用*标注