引言

Join是SQL语句中的常用操作,良好的表结构能够将数据分散在不同的表中,使其符合某种范式,减少表冗余、更新容错等。而建立表和表之间关系的最佳方式就是Join操作。

对于Spark来说有3中Join的实现,每种Join对应着不同的应用场景:

  1. Broadcast Hash Join : 适合一张较小的表和一张大表进行join
  2. Shuffle Hash Join :  适合一张小表和一张大表进行join,或者是两张小表之间的join
  3. Sort Merge Join : 适合两张较大的表之间进行join

前两者都基于的是Hash Join,只不过在hash join之前需要先shuffle还是先broadcast。下面将详细的解释一下这三种不同的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在一起。

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

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

大家知道,在数据库的常见模型中(比如星型模型或者雪花模型),表一般分为两种:事实表和维度表。维度表一般指固定的、变动较少的表,例如联系人、物品种类等,一般数据有限。而事实表一般记录流水,比如销售清单等,通常随着时间的增长不断膨胀。

因为Join操作是对两个表中key值相同的记录进行连接,在SparkSQL中,对两个表做Join最直接的方式是先根据key分区,再在每个分区中把key值相同的记录拿出来做连接操作。但这样就不可避免地涉及到shuffle,而shuffle在Spark中是比较耗时的操作,我们应该尽可能的设计Spark应用使其避免大量的shuffle。

当维度表和事实表进行Join操作时,为了避免shuffle,我们可以将大小有限的维度表的全部数据分发到每个节点上,供事实表使用。executor存储维度表的全部数据,一定程度上牺牲了空间,换取shuffle操作大量的耗时,这在SparkSQL中称作Broadcast Join,如下图所示:

Table B是较小的表,黑色表示将其广播到每个executor节点上,Table A的每个partition会通过block manager取到Table A的数据。根据每条记录的Join Key取到Table B中相对应的记录,根据Join Type进行操作。这个过程比较简单,不做赘述。

Broadcast Join的条件有以下几个:

1. 被广播的表需要小于spark.sql.autoBroadcastJoinThreshold所配置的值,默认是10M (或者加了broadcast join的hint)

2. 基表不能被广播,比如left outer join时,只能广播右表

看起来广播是一个比较理想的方案,但它有没有缺点呢?也很明显。这个方案只能用于广播较小的表,否则数据的冗余传输就远大于shuffle的开销;另外,广播时需要将被广播的表现collect到driver端,当频繁有广播出现时,对driver的内存也是一个考验。

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

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

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

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

Shuffle Hash Join

当一侧的表比较小时,我们选择将其广播出去以避免shuffle,提高性能。但因为被广播的表首先被collect到driver段,然后被冗余分发到每个executor上,所以当表比较大时,采用broadcast join会对driver端和executor端造成较大的压力。

但由于Spark是一个分布式的计算引擎,可以通过分区的形式将大批量的数据划分成n份较小的数据集进行并行计算。这种思想应用到Join上便是Shuffle Hash Join了。利用key相同必然分区相同的这个原理,两个表中,key相同的行都会被shuffle到同一个分区中,SparkSQL将较大表的join分而治之,先将表划分成n个分区,再对两个表中相对应分区的数据分别进行Hash Join,这样即在一定程度上减少了driver广播一侧表的压力,也减少了executor端取整张被广播表的内存消耗。其原理如下图:

Shuffle Hash Join分为两步:

1. 对两张表分别按照join keys进行重分区,即shuffle,目的是为了让有相同join keys值的记录分到对应的分区中

2. 对对应分区中的数据进行join,此处先将小表分区构造为一张hash表,然后根据大表分区中记录的join keys值拿出来进行匹配

Shuffle Hash Join的条件有以下几个:

1. 分区的平均大小不超过spark.sql.autoBroadcastJoinThreshold所配置的值,默认是10M

2. 基表不能被广播,比如left outer join时,只能广播右表

3. 一侧的表要明显小于另外一侧,小的一侧将被广播(明显小于的定义为3倍小,此处为经验值)

我们可以看到,在一定大小的表中,SparkSQL从时空结合的角度来看,将两个表进行重新分区,并且对小表中的分区进行hash化,从而完成join。在保持一定复杂度的基础上,尽量减少driver和executor的内存压力,提升了计算时的稳定性。

在大数据条件下如果一张表很小,执行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算法。

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

Sort Merge Join

上面介绍的两种实现对于一定大小的表比较适用,但当两个表都非常大时,显然无论适用哪种都会对计算内存造成很大压力。这是因为join时两者采取的都是hash join,是将一侧的数据完全加载到内存中,使用hash code取join keys值相等的记录进行连接。

当两个表都非常大时,SparkSQL采用了一种全新的方案来对表进行Join,即Sort Merge Join。这种实现方式不用将一侧数据全部加载后再进星hash join,但需要在join前将数据排序,如下图所示:

可以看到,首先将两张表按照join keys进行了重新shuffle,保证join keys值相同的记录会被分在相应的分区。分区后对每个分区内的数据进行排序,排序后再对相应的分区内的记录进行连接,如下图示:

看着很眼熟吧?也很简单,因为两个序列都是有序的,从头遍历,碰到key相同的就输出;如果不同,左边小就继续取左边,反之取右边。

可以看出,无论分区有多大,Sort Merge Join都不用把某一侧的数据全部加载到内存中,而是即用即取即丢,从而大大提升了大数据量下sql join的稳定性。

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

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

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

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

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

参考来源:

https://www.cnblogs.com/0xcafedaddy/p/7614299.html

转载于:https://www.cnblogs.com/JP6907/p/10721436.html

SparkSQL的3种Join实现相关推荐

  1. Oracle的join默认为,Oracle中的三种Join方法详解

    这里将为大家介绍Oracle中的三种Join方法,Nested loop join.Sort merge join和Hash join.整理出来以便帮助大家学习. 基本概念 Nested loop j ...

  2. oracle hash join outer,CSS_浅谈Oracle中的三种Join方法,基本概念 Nested loop join: Outer - phpStudy...

    浅谈Oracle中的三种Join方法 基本概念 Nested loop join: Outer table中的每一行与inner table中的相应记录join,类似一个嵌套的循环. Sort mer ...

  3. 061 hive中的三种join与数据倾斜

    一:hive中的三种join 1.map join 应用场景:小表join大表 一:设置mapjoin的方式: )如果有一张表是小表,小表将自动执行map join. 默认是true. <pro ...

  4. 图解 5 种 Join 连接及实战案例!(inner/ left/ right/ full/ cross)

    转载自 图解 5 种 Join 连接及实战案例!(inner/ left/ right/ full/ cross) Join 连接在日常开发用得比较多,但大家都搞清楚了它们的使用区别吗??一文带你上车 ...

  5. left join 多条件_第九篇|Spark的五种JOIN策略解析

    JOIN操作是非常常见的数据处理操作,Spark作为一个统一的大数据处理引擎,提供了非常丰富的JOIN场景.本文分享将介绍Spark所提供的5种JOIN策略,希望对你有所帮助.本文主要包括以下内容: ...

  6. MapReduce三种join实例分析

    本文引自吴超博客 实现原理 1.在Reudce端进行连接. 在Reudce端进行连接是MapReduce框架进行表之间join操作最为常见的模式,其具体的实现原理如下: Map端的主要工作:为来自不同 ...

  7. sparksql中shuffled hash join策略

    背景 sparksql中又很多join策略,其中有一个是shuffled hash join.这个用的比较多,但是又说不明白其原理,我各种百度,各种谷歌,最后在stackoverflow找到了一份资料 ...

  8. MySQL的两张表的七种Join查询

    SQL的语法格式如下 SELECT DISTINCT< select_list > FROM< left_table > < join_type > JOIN &l ...

  9. MySQL的7种join,交集,并集,差集。

    MySQL的7种join,交集,并集,差集. 1.总括 mysql种共有7种join(但实际上只有4种主要的:左外连接,右外连接,内连接,全外连接)以下7种不过在此基础上有一些变动: 2.具体实现 ( ...

最新文章

  1. identifyTask练习
  2. 笔记-项目沟通管理-规划沟通管理
  3. bin文件用cad打开_dwg文件怎么打开?CAD看图,360°精确识别CAD图块,细节见真章...
  4. windows mobile创建文本文件并用word打开
  5. 我如何构建Kubernetes集群,以便我的同事可以更快地部署应用程序
  6. js基础教程学习笔记
  7. 富士康有麻烦了,iPhone产能已被削减
  8. SQL Server执行计划面试问题
  9. qt web混合编程_基于Qt与MATLAB的混合编程技术
  10. UE4 讯飞语音识别插件
  11. 实验过程中收获的经验、教训、感想
  12. dimens文件生成器使用方法
  13. Dva员工增删改查Demo实现-优化
  14. 关于计算机的英语作文300字,关于电脑的英语作文集锦5篇
  15. 前端职业规划 - 写给年轻的前端韭菜们
  16. 01_CCC3.0数字钥匙_蓝牙OOB配对过程
  17. Linux设备模型(1)_基本概念
  18. 粉末成型工艺(粉末冶金粉末注射成型)
  19. python爬取豆瓣top250信息并存入数据库中 | sqlite3
  20. totolinkn200up怎么设置_TOTOLINK EX300无线中继器设置说明

热门文章

  1. linux长期版本维护内容,[图]Linux Kernel 4.20首个维护版本更新发布 已稳定可广泛部署...
  2. java自定义返回码_java – 自定义HTTP状态代码
  3. 软件工程实训有必要吗_软件工程实训报告的总结.docx
  4. 计算机操作系统专题一:多道环境下进程同步与互斥制约关系的学习
  5. esp32cam与下载板的实际有效接线图
  6. docker 启动容器_crontab入门二:定时启动docker容器并启动容器内部脚本
  7. java栈代码执行_这段实现栈的java代码执行错误..
  8. nginx 转发慢_学习Nginx的正确姿势,多图详解助你更上一层楼!(干货收藏篇)...
  9. TCP/IP 协议讲的明明白白了
  10. setting an array element with a sequence.