0
  • 聊天消息
  • 系统消息
  • 评论与回复
登录后你可以
  • 下载海量资料
  • 学习在线课程
  • 观看技术视频
  • 写文章/发帖/加入社区
创作中心

完善资料让更多小伙伴认识你,还能领取20积分哦,立即完善>

3天内不再提示

Join在Spark中是如何组织运行的

人工智能与大数据技术 来源:人工智能与大数据技术 作者:人工智能与大数据 2020-09-25 11:35 次阅读

Join作为SQL中一个重要语法特性,几乎所有稍微复杂一点的数据分析场景都离不开Join,如今Spark SQL(Dataset/DataFrame)已经成为Spark应用程序开发的主流,作为开发者,我们有必要了解Join在Spark中是如何组织运行的。

SparkSQL总体流程介绍

在阐述Join实现之前,我们首先简单介绍SparkSQL的总体流程,一般地,我们有两种方式使用SparkSQL,一种是直接写sql语句,这个需要有元数据库支持,例如Hive等,另一种是通过Dataset/DataFrame编写Spark应用程序。如下图所示,sql语句被语法解析(SQL AST)成查询计划,或者我们通过Dataset/DataFrame提供的APIs组织成查询计划,查询计划分为两大类:逻辑计划和物理计划,这个阶段通常叫做逻辑计划,经过语法分析(Analyzer)、一系列查询优化(Optimizer)后得到优化后的逻辑计划,最后被映射成物理计划,转换成RDD执行。

对于语法解析、语法分析以及查询优化,本文不做详细阐述,本文重点介绍Join的物理执行过程。

Join基本要素

如下图所示,Join大致包括三个要素:Join方式、Join条件以及过滤条件。其中过滤条件也可以通过AND语句放在Join条件中。

Spark支持所有类型的Join,包括:

inner join

left outer join

right outer join

full outer join

left semi join

left anti join

下面分别阐述这几种Join的实现。

Join基本实现流程

总体上来说,Join的基本实现流程如下图所示,Spark将参与Join的两张表抽象为流式遍历表(streamIter)和查找表(buildIter),通常streamIter为大表,buildIter为小表,我们不用担心哪个表为streamIter,哪个表为buildIter,这个spark会根据join语句自动帮我们完成。

在实际计算时,spark会基于streamIter来遍历,每次取出streamIter中的一条记录rowA,根据Join条件计算keyA,然后根据该keyA去buildIter中查找所有满足Join条件(keyB==keyA)的记录rowBs,并将rowBs中每条记录分别与rowAjoin得到join后的记录,最后根据过滤条件得到最终join的记录。

从上述计算过程中不难发现,对于每条来自streamIter的记录,都要去buildIter中查找匹配的记录,所以buildIter一定要是查找性能较优的数据结构。spark提供了三种join实现:sort merge join、broadcast join以及hash join。

sort merge join实现

要让两条记录能join到一起,首先需要将具有相同key的记录在同一个分区,所以通常来说,需要做一次shuffle,map阶段根据join条件确定每条记录的key,基于该key做shuffle write,将可能join到一起的记录分到同一个分区中,这样在shuffle read阶段就可以将两个表中具有相同key的记录拉到同一个分区处理。前面我们也提到,对于buildIter一定要是查找性能较优的数据结构,通常我们能想到hash表,但是对于一张较大的表来说,不可能将所有记录全部放到hash表中,另外也可以对buildIter先排序,查找时按顺序查找,查找代价也是可以接受的,我们知道,spark shuffle阶段天然就支持排序,这个是非常好实现的,下面是sort merge join示意图。

在shuffle read阶段,分别对streamIter和buildIter进行merge sort,在遍历streamIter时,对于每条记录,都采用顺序查找的方式从buildIter查找对应的记录,由于两个表都是排序的,每次处理完streamIter的一条记录后,对于streamIter的下一条记录,只需从buildIter中上一次查找结束的位置开始查找,所以说每次在buildIter中查找不必重头开始,整体上来说,查找性能还是较优的。

broadcast join实现

为了能具有相同key的记录分到同一个分区,我们通常是做shuffle,那么如果buildIter是一个非常小的表,那么其实就没有必要大动干戈做shuffle了,直接将buildIter广播到每个计算节点,然后将buildIter放到hash表中,如下图所示。

从上图可以看到,不用做shuffle,可以直接在一个map中完成,通常这种join也称之为map join。那么问题来了,什么时候会用broadcast join实现呢?这个不用我们担心,spark sql自动帮我们完成,当buildIter的估计大小不超过参数spark.sql.autoBroadcastJoinThreshold设定的值(默认10M),那么就会自动采用broadcast join,否则采用sort merge join。

hash join实现

除了上面两种join实现方式外,spark还提供了hash join实现方式,在shuffle read阶段不对记录排序,反正来自两格表的具有相同key的记录会在同一个分区,只是在分区内不排序,将来自buildIter的记录放到hash表中,以便查找,如下图所示。

不难发现,要将来自buildIter的记录放到hash表中,那么每个分区来自buildIter的记录不能太大,否则就存不下,默认情况下hash join的实现是关闭状态,如果要使用hash join,必须满足以下四个条件:

buildIter总体估计大小超过spark.sql.autoBroadcastJoinThreshold设定的值,即不满足broadcast join条件

开启尝试使用hash join的开关,spark.sql.join.preferSortMergeJoin=false

每个分区的平均大小不超过spark.sql.autoBroadcastJoinThreshold设定的值,即shuffle read阶段每个分区来自buildIter的记录要能放到内存中

streamIter的大小是buildIter三倍以上

所以说,使用hash join的条件其实是很苛刻的,在大多数实际场景中,即使能使用hash join,但是使用sort merge join也不会比hash join差很多,所以尽量使用hash

下面我们分别阐述不同Join方式的实现流程。

inner join

inner join是一定要找到左右表中满足join条件的记录,我们在写sql语句或者使用DataFrame时,可以不用关心哪个是左表,哪个是右表,在spark sql查询优化阶段,spark会自动将大表设为左表,即streamIter,将小表设为右表,即buildIter。这样对小表的查找相对更优。其基本实现流程如下图所示,在查找阶段,如果右表不存在满足join条件的记录,则跳过。

left outer join

left outer join是以左表为准,在右表中查找匹配的记录,如果查找失败,则返回一个所有字段都为null的记录。我们在写sql语句或者使用DataFrmae时,一般让大表在左边,小表在右边。其基本实现流程如下图所示。

right outer join

right outer join是以右表为准,在左表中查找匹配的记录,如果查找失败,则返回一个所有字段都为null的记录。所以说,右表是streamIter,左表是buildIter,我们在写sql语句或者使用DataFrame时,一般让大表在右边,小表在左边。其基本实现流程如下图所示。

full outer join

full outer join相对来说要复杂一点,总体上来看既要做left outer join,又要做right outer join,但是又不能简单地先left outer join,再right outer join,最后union得到最终结果,因为这样最终结果中就存在两份inner join的结果了。因为既然完成left outer join又要完成right outer join,所以full outer join仅采用sort merge join实现,左边和右表既要作为streamIter,又要作为buildIter,其基本实现流程如下图所示。

由于左表和右表已经排好序,首先分别顺序取出左表和右表中的一条记录,比较key,如果key相等,则joinrowA和rowB,并将rowA和rowB分别更新到左表和右表的下一条记录;如果keyAkeyB,则说明左表中没有与右表rowB对应的记录,那么joinnullRow与rowB,紧接着,rowB更新到右表的下一条记录。如此循环遍历直到左表和右表的记录全部处理完。

left semi join

left semi join是以左表为准,在右表中查找匹配的记录,如果查找成功,则仅返回左边的记录,否则返回null,其基本实现流程如下图所示。

left anti join

left anti join与left semi join相反,是以左表为准,在右表中查找匹配的记录,如果查找成功,则返回null,否则仅返回左边的记录,其基本实现流程如下图所示。

总结

Join是数据库查询中一个非常重要的语法特性,在数据库领域可以说是“得join者得天下”,SparkSQL作为一种分布式数据仓库系统,给我们提供了全面的join支持,并在内部实现上无声无息地做了很多优化,了解join的实现将有助于我们更深刻的了解我们的应用程序的运行轨迹。

责任编辑:xj

原文标题:面试必知的 Spark SQL 几种 Join 实现

文章出处:【微信公众号:人工智能与大数据技术】欢迎添加关注!文章转载请注明出处。

声明:本文内容及配图由入驻作者撰写或者入驻合作网站授权转载。文章观点仅代表作者本人,不代表电子发烧友网立场。文章及其配图仅供工程师学习之用,如有内容侵权或者其他违规问题,请联系本站处理。 举报投诉
  • SQL
    SQL
    +关注

    关注

    1

    文章

    734

    浏览量

    43422
  • SPARK
    +关注

    关注

    1

    文章

    98

    浏览量

    19707

原文标题:面试必知的 Spark SQL 几种 Join 实现

文章出处:【微信号:TheBigData1024,微信公众号:人工智能与大数据技术】欢迎添加关注!文章转载请注明出处。

收藏 人收藏

    评论

    相关推荐

    STM32HIAR如何实现从FLASH加载到SRAM运行程序?

    如题,STM32H IAR如何实现从FLASH加载到SRAM运行程序 有没有相关的例程可供参考
    发表于 03-28 07:46

    Spark基于DPU Snappy压缩算法的异构加速方案

    一、总体介绍 1.1 背景介绍 Apache Spark是专为大规模数据计算而设计的快速通用的计算引擎,是一种与 Hadoop 相似的开源集群计算环境,但是两者之间还存在一些不同之处,这些不同之处
    的头像 发表于 03-26 17:06 126次阅读
    <b class='flag-5'>Spark</b>基于DPU Snappy压缩算法的异构加速方案

    RDMA技术在Apache Spark中的应用

    背景介绍 在当今数据驱动的时代,Apache Spark已经成为了处理大规模数据集的首选框架。作为一个开源的分布式计算系统,Spark因其高效的大数据处理能力而在各行各业中广受欢迎。无论是金融服务
    的头像 发表于 03-25 18:13 901次阅读
    RDMA技术在Apache <b class='flag-5'>Spark</b>中的应用

    基于DPU和HADOS-RACE加速Spark 3.x

    背景简介 Apache Spark(下文简称Spark)是一种开源集群计算引擎,支持批/流计算、SQL分析、机器学习、图计算等计算范式,以其强大的容错能力、可扩展性、函数式API、多语言支持(SQL
    的头像 发表于 03-25 18:12 849次阅读
    基于DPU和HADOS-RACE加速<b class='flag-5'>Spark</b> 3.x

    MySQL到底是join性能好,还是in一下更快呢?

    事情是这样的,去年入职的新公司,之后在代码review的时候被提出说,不要写joinjoin耗性能还是慢来着,当时也是真的没有多想,那就写in好了,最近发现in的数据量过大的时候会导致sql慢,甚至sql太长,直接报错了。
    的头像 发表于 10-31 15:40 161次阅读
    MySQL到底是<b class='flag-5'>join</b>性能好,还是in一下更快呢?

    KEIL如何让程序RAM运行

    STM32F411Nucleo 的一个例子来介绍几种让程序 RAM 运行的方法。该例子,通过调用 ToggleLED 函数来翻转
    发表于 09-28 08:19

    IAR下如何将程序拷贝RAM运行

    有时候我们需要将一部分程序从 FLASH 拷贝到 RAM 运行,以提高程序运行的速度。本文基于 IAREmbedded Workbench 列出了一些方法,可以系统启动时自动从 F
    发表于 09-28 06:50

    NVIDIA 携手腾讯开发和优化 Spark UCX 实现性能跃升

    腾讯网络平台部与数据平台部,联合 NVIDIA 合作开发和优化 Spark UCX,最终实现 Spark Shuffle 稳定加速 15% - 20%,平均降低现网 Spark 任务
    的头像 发表于 08-25 20:50 398次阅读
    NVIDIA 携手腾讯开发和优化 <b class='flag-5'>Spark</b> UCX 实现性能跃升

    你还在手写join联表查询?MyBatis-Plus这样写太香了!

    众所周知,mybatis plus 封装的 mapper 不支持 join,如果需要支持就必须自己去实现。但是对于大部分的业务场景来说,都需要多表 join,要不然就没必要采用关系型数据库了。
    的头像 发表于 07-07 10:19 771次阅读
    你还在手写<b class='flag-5'>join</b>联表查询?MyBatis-Plus这样写太香了!

    Spark ML算法提供GPU加速度

    新的 GPU 库降低了 Apache Spark ML 的计算成本
    的头像 发表于 07-05 16:30 539次阅读
    为<b class='flag-5'>Spark</b> ML算法提供GPU加速度

    Spark 3.4用于分布式模型训练和大规模模型推理

    使用 Spark 3.4 简化分布式深度学习
    的头像 发表于 07-05 16:30 718次阅读
    <b class='flag-5'>Spark</b> 3.4用于分布式模型训练和大规模模型推理

    在使用left jion时,on和where条件的区别

    sql :   select * from student s left join class c on s.classId=c.id order by s.id   运行sql :   select * from student s left 
    的头像 发表于 06-16 14:46 333次阅读
    在使用left jion时,on和where条件的区别

    数据治理组织架构如何有效运行和落地?_光点科技

    在当今数字化的时代,数据成为了企业重要的资产之一,而数据治理是保证数据质量和有效性的基础。一个有效的数据治理组织架构能够提升企业的数据价值,并确保合规性和风险控制。但是,如何建立一个能够有效运行和落地的数据治理组织架构是一个复杂
    的头像 发表于 05-06 10:27 395次阅读

    如何优化MySQL中的join语句

    在mysql中,join 主要有Nested Loop、Hash Join、Merge Join 这三种方式,我们今天来看一下最普遍 Nested Loop 循环连接方式,主要包括三种:
    的头像 发表于 04-24 17:03 537次阅读
    如何优化MySQL中的<b class='flag-5'>join</b>语句

    LLCE driver是否可以非autosar os运行

    S32DS、RTD和LLCE CAN驱动有对应的工具版本吗?我想知道,LLCE driver是否可以非autosar os运行,是否有关于LLCE Driver +S32DS3.5的文档可以学习。
    发表于 04-18 09:21