一、 数据准备
本文主要介绍 Spark SQL 的多表连接,需要预先准备测试数据。分别创建员工和部门的 Datafame,并注册为临时视图,代码如下:
val spark = SparkSession.builder().appName("aggregations").master("local[2]").getOrCreate()
val empDF = spark.read.json("/usr/file/json/emp.json")
empDF.createOrReplaceTempView("emp")
val deptDF = spark.read.json("/usr/file/json/dept.json")
deptDF.createOrReplaceTempView("dept")
两表的主要字段如下:
emp 员工表 |-- ENAME: 员工姓名 |-- DEPTNO: 部门编号 |-- EMPNO: 员工编号 |-- HIREDATE: 入职时间 |-- JOB: 职务 |-- MGR: 上级编号 |-- SAL: 薪资 |-- COMM: 奖金
dept 部门表 |-- DEPTNO: 部门编号 |-- DNAME: 部门名称 |-- LOC: 部门所在城市
注:emp.json,dept.json 可以在本仓库的resources 目录进行下载。
二、连接类型
Spark 中支持多种连接类型:
- Inner Join : 内连接;
- Full Outer Join : 全外连接;
- Left Outer Join : 左外连接;
- Right Outer Join : 右外连接;
- Left Semi Join : 左半连接;
- Left Anti Join : 左反连接;
- Natural Join : 自然连接;
- Cross (or Cartesian) Join : 交叉 (或笛卡尔) 连接。
其中内,外连接,笛卡尔积均与普通关系型数据库中的相同,如下图所示:
这里解释一下左半连接和左反连接,这两个连接等价于关系型数据库中的in和not in字句:
-- LEFT SEMI JOIN
SELECT * FROM emp LEFT SEMI JOIN dept ON emp.deptno = dept.deptno
-- 等价于如下的 IN 语句
SELECT * FROM emp WHERE deptno IN (SELECT deptno FROM dept)
-- LEFT ANTI JOIN
SELECT * FROM emp LEFT ANTI JOIN dept ON emp.deptno = dept.deptno
-- 等价于如下的 IN 语句
SELECT * FROM emp WHERE deptno NOT IN (SELECT deptno FROM dept)
所有连接类型的示例代码如下:
2.1 inner join
两表内接
// 1.定义连接表达式
val joinExpression = empDF.col("deptno") === deptDF.col("deptno")
// 2.连接查询
empDF.join(deptDF,joinExpression).select("ename","dname").show()
// 等价 SQL 如下:
spark.sql("SELECT ename,dname FROM emp JOIN dept ON emp.deptno = dept.deptno").show()
2.2 full outer join
FULL OUTER JOIN 关键字返回左表(Websites)和右表(access_log)中所有的行。
empDF.join(deptDF, joinExpression, "outer").show()
spark.sql("SELECT * FROM emp FULL OUTER JOIN dept ON emp.deptno = dept.deptno").show()
2.3 left outer join
把左边表的数据全部取出来,而右边表的数据有相等的,显示出来,如果没有,显示NULL
empDF.join(deptDF, joinExpression, "left_outer").show()
spark.sql("SELECT * FROM emp LEFT OUTER JOIN dept ON emp.deptno = dept.deptno").show()
2.4 right outer join
把右边表的数据全部取出来,而左边表的数据有相等的,显示出来,如果没有,显示NULL
empDF.join(deptDF, joinExpression, "right_outer").show()
spark.sql("SELECT * FROM emp RIGHT OUTER JOIN dept ON emp.deptno = dept.deptno").show()
2.5 left_semi join
Semi Join,也叫半连接,是从分布式数据库中借鉴过来的方法。它的产生动机是:对于reduce side join,跨机器的数据传输量非常大,这成了join操作的一个瓶颈,如果能够在map端过滤掉不会参加join操作的数据,则可以大大节省网络IO,提升执行效率。
left_semi join子句中右边的表只能在 ON 子句中设置过滤条件,在 WHERE 子句、SELECT 子句或其他地方过滤都不行。
left_semi join和join对待右表中重复key的处理方式差异:因为 left semi join 是 in(keySet) 的关系,遇到右表重复记录,左表会跳过,而 join on 则会一直遍历。
最后的结果是这会造成性能,以及 join 结果上的差异。
left semi join 中最后 select 的结果只许出现左表,因为右表只有 join key 参与关联计算了,而 join on 默认是整个关系模型都参与计算了。
empDF.join(deptDF, joinExpression, "left_semi").show()
spark.sql("SELECT * FROM emp LEFT SEMI JOIN dept ON emp.deptno = dept.deptno").show()
2.6 left anti join
left anti join的功能是在查询过程中,剔除左表中和右表有交集的部分
empDF.join(deptDF, joinExpression, "left_anti").show()
spark.sql("SELECT * FROM emp LEFT ANTI JOIN dept ON emp.deptno = dept.deptno").show()
2.7 cross join
CROSS JOIN 称为“交叉连接”或者“笛卡尔连接”。SQL CROSS JOIN 连接用于从两个或者多个连接表中返回记录集的笛卡尔积,即将左表的每一行与右表的每一行合并。
empDF.join(deptDF, joinExpression, "cross").show()
spark.sql("SELECT * FROM emp CROSS JOIN dept ON emp.deptno = dept.deptno").show()
2.8 natural join
自然连接是在两张表中寻找那些数据类型和列名都相同的字段,然后自动地将他们连接起来,并返回所有符合条件的结果。
spark.sql("SELECT * FROM emp NATURAL JOIN dept").show()
以下是一个自然连接的查询结果,程序自动推断出使用两张表都存在的 dept 列进行连接,其实际等价于:
spark.sql("SELECT * FROM emp JOIN dept ON emp.deptno = dept.deptno").show()
三、连接的执行
在对大表与大表之间进行连接操作时,通常都会触发shuffle join,两表的所有分区节点会进行ALL-to-ALL的通讯,这种查询通常比较昂贵,会对网络 IO 会造成比较大的负担。
而对于大表和小表的连接操作,Spark 会在一定程度上进行优化,如果小表的数据量小于 Worker Node 的内存空间,Spark 会考虑将小表的数据广播到每一个 Worker Node,在每个工作节点内部执行连接计算,这可以降低网络的 IO,但会加大每个 Worker Node 的 CPU 负担。
文章来源:https://www.toymoban.com/news/detail-414723.html
是否采用广播方式进行 Join
取决于程序内部对小表的判断,如果想明确使用广播方式进行 Join
,则可以在 DataFrame API 中使用 broadcast
方法指定需要广播的小表:文章来源地址https://www.toymoban.com/news/detail-414723.html
empDF.join(broadcast(deptDF), joinExpression).show()
到了这里,关于Spark SQL join操作详解的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!