title: Spark SQL JOIN操作
date: 2021-05-08 15:53:21
tags:

Spark

本文主要介绍 Spark SQL 的多表连接，需要预先准备测试数据。分别创建员工和部门的 Datafame，并注册为临时视图.

一、数据准备

本文主要介绍 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

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

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

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

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

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

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 负担。

是否采用广播方式进行 Join 取决于程序内部对小表的判断，如果想明确使用广播方式进行 Join，则可以在 DataFrame API 中使用 broadcast 方法指定需要广播的小表：

empDF.join(broadcast(deptDF), joinExpression).show()

参考链接：

https://blog.csdn.net/m0_37809146/article/details/91282446

Spark SQL JOIN操作代码示例相关推荐

Spark Checkpoint读操作代码分析
上次介绍了RDD的Checkpint写过程(<Spark Checkpoint写操作代码分析>),本文将介绍RDD如何读取已经Checkpoint的数据.在RDD Checkpoint完之 ...
Spark SQL程序操作HiveContext
HiveContext是对应spark-hive这个项目,与hive有部分耦合, 支持hql,是SqlContext的子类,也就是说兼容SqlContext; 1.添加依赖 <dependenc ...
Spark SQL join的三种实现方式
引言 join是SQL中的常用操作,良好的表结构能够将数据分散到不同的表中,使其符合某种规范(mysql三大范式),可以最大程度的减少数据冗余,更新容错等,而建立表和表之间关系的最佳方式就是join操 ...
【约束布局】ConstraintLayout 屏障 Barrier 约束 ( 简介 | 屏障适用场景 | 位置说明 | 相关属性 | 可视化操作 | 代码示例 | 总结 )
文章目录 I . 屏障 Barrier 约束简介 II . 屏障线 Barrier 与引导线 Guideline 对比 III . 屏障 Barrier 适用场景 IV . 屏障 Barrier ...
java ldap 工具_通过定义LdapUtil工具类实现JAVA关于LDAP协议目录数据保存和删除等操作代码示例...
一.前言关于基于jdk开发包javax.naming.directory.DirContext.javax.naming.directory.InitialDirContext目录协议定义LdapU ...
Android文件操作代码示例,sd卡数据储存
1. public class MainActivity extends AppCompatActivity {@Overrideprotected void onCreate(Bundle save ...
小程序获取城市行政区号_小程序获取用户手机号操作代码示例
小程序获取手机号操作,获取微信用户绑定的手机号,需先调用wx.login接口.因为需要用户主动触发才能发起获取手机号接口,所以该功能不由 API 来调用,需用组件的点击来触发,也就是用户主动点击按钮 ...
Python3字符串切片操作代码示例
Spark详解（十四）：Spark SQL的Join实现
1. 简介 Join是SQL语句中的常用操作,良好的表结构能够将数据分散在不同的表中,使其符合某种范式,减少表冗余.更新容错等.而建立表和表之间关系的最佳方式就是Join操作.Join连接是大数据处理 ...

Spark SQL JOIN操作代码示例