AI 助理
备案控制台
开发者社区 数据库 文章 正文

【Spark】（八）Spark SQL 应用解析1

2022-06-10 124
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
全局流量管理 GTM，标准版 1个月
公共DNS（含HTTPDNS解析），每月1000万次HTTP解析
云解析 DNS，旗舰版 1个月
简介： 【Spark】（八）Spark SQL 应用解析1

文章目录

一、Spark SQL的进化之路


二、认识Spark SQL

2.1　什么是Spark SQL?

2.2　Spark SQL的作用

2.3　运行原理

2.4　特点

2.5　Spark SQL数据抽象


三、Spark SQL API

3.1　SparkSession


3.2　DataSet ( Spark1. 6 + )

1、创建 DataSet

2、使用case Class 创建 DataSet

3、使用DataSet完成Demo


3.3　DataFrame

1、创建 DataFrame

2、使用DataFrame完成Demo


四、Spark SQL 操作Hive表

4.1　文件配置

4.1　操作 Hive 表


五、Spark SQL 连 MySQL

六、Spark SQL 内置函数

七、Spark SQL 自定义函数


一、Spark SQL的进化之路

1.0以前： Shark


1.1：x开始：Spark SQL(只是测试性的) SQL


1.3.x：Spark SQL(正式版本)+Dataframe


1.5.x：Spark SQL 钨丝计划

b

1.6.x：Spark SQL+DataFrame+DataSet(测试版本)


x：Spark SQL+DataFrame+DataSet(正式版本)

Spark SQL：还有其他的优化 StructuredStreaming(DataSet)


二、认识Spark SQL

2.1　什么是Spark SQL?


spark SQL是spark的一个模块，主要用于进行结构化数据的处理。它提供的最核心的编程抽象就是DataFrame。


image.png


2.2　Spark SQL的作用


提供一个编程抽象（DataFrame） 并且作为分布式 SQL 查询引擎


DataFrame：它可以根据很多源进行构建，包括：结构化的数据文件，hive中的表，外部的关系型数据库，以及RDD


2.3　运行原理


将 Spark SQL 转化为 RDD， 然后提交到集群执行


2.4　特点


（1）容易整合


（2）统一的数据访问方式


（3）兼容 Hive


（4）标准的数据连接


2.5　Spark SQL数据抽象


RDD (Spark1.0) -> DataFrame (Spark1.3) -> DataSet ( Spark1. 6)


  • Spark SQL提供了DataFrame和DataSet的数据抽象。
  • DataFrame就是RDD + Schema,可以认为是一-张二维表格。 他的劣势是在编译器不进行表格中的字段的类型检查。在运行期进行检查。
  • DataSet是 Spark最新的数据抽象, Spark的发 展会逐步将DataSet作为主要的数据抽象，弱化RDD和DataFrame。 DataSet包含了DataFr ame所有的优化机制。除此之外提供了以样例类为Schema模型的强类型。
  • DataFrame = DataSet [Row]
  • DataFrame和DataSet都有可控的内存管理机制，所有数据都保存在非堆上，都使用了catalyst进行SQL的优化。


image.png


三、Spark SQL API

3.1　SparkSession

  • SQLContext Spark SQL的编程入口
  • HiveContext的子集，包含更多的功能
  • SparkSession (Spark 2.x 推荐)
  • SparkSession：合并了SQLContext与HiveContext
  • 提供与Spark功能交互单一入口点，并允许使用DataFrame和DataSet API对Spark进行编程


SparkSession 是 Spark 2.0 引如的新概念。SparkSession为用户提供了统一的切入点，来让用户学习spark的各项功能。 在spark的早期版本中，SparkContext 是 spark 的主要切入点，由于RDD是主要的API，我们通过sparkcontext来创建和操作RDD。对于每个其他的API，我们需要使用不同的context。例如，对于Streming，我们需要使用 StreamingContext ；对于sql，使用 sqlContext ；对于Hive，使用 hiveContext 。但是随着 DataSet 和 DataFrame 的API逐渐成为标准的API，就需要为他们建立接入点。所以在spark2.0中，引入SparkSession作为DataSet和DataFrame API 的切入点，SparkSession封装了SparkConf、SparkContext和SQLContext。为了向后兼容，SQLContext和HiveContext也被保存下来。


SparkSession 实质上是 SQLContext 和 HiveContext 的组合（未来可能还会加上 StreamingContext），所以在 SQLContext 和 HiveContext 上可用的 API 在 SparkSession 上同样是可以使用的。SparkSession 内部封装了 SparkContext，所以计算实际上是由 SparkContext 完成的。


特点：


  • 为用户提供一个统一的切入点使用Spark 各项功能


  • 允许用户通过它调用 DataFrame 和 Dataset 相关 API 来编写程序


  • 减少了用户需要了解的一些概念，可以很容易的与 Spark 进行交互


  • 与 Spark 交互之时不需要显示的创建 SparkConf, SparkContext 以及 SQlContext，这些对象已经封闭在 SparkSession 中


3.2　DataSet ( Spark1. 6 + )


image.png


1、创建 DataSet

createDataset（）的参数可以是：Seq、Array、RDD
scala> spark.createDataset(1 to 3).show
+-----+
|value|
+-----+
|    1|
|    2|
|    3|
+-----+
scala> spark.createDataset(List(("a",1))).show
+---+---+
| _1| _2|
+---+---+
|  a|  1|
+---+---+


2、使用case Class 创建 DataSet


Scala中在class 关键字前面加上case关键字，这个类就可以成为样例类

scala> case class Userinfos(userid:Int,username:String)
defined class Userinfos
scala> case class Scores(sid:Int,userid:Int,score:Int)
defined class Scores
scala> val users = Seq(Userinfos(1,"zhangsan"),Userinfos(2,"lisi"),Userinfos(3,"Wangwu")).toDS
users: org.apache.spark.sql.Dataset[Userinfos] = [userid: int, username: string]
scala> val scs = Seq(Scores(1,1,100),Scores(2,1,30),Scores(3,2,60),Scores(4,3,35)).toDS
scs: org.apache.spark.sql.Dataset[Scores] = [sid: int, userid: int ... 1 more field]
scala> users.join(scs,users("userid")===scs("userid")).show
+------+--------+---+------+-----+
|userid|username|sid|userid|score|
+------+--------+---+------+-----+
|     1|zhangsan|  2|     1|   30|
|     1|zhangsan|  1|     1|  100|
|     2|    lisi|  3|     2|   60|
|     3|  Wangwu|  4|     3|   35|
+------+--------+---+------+-----+


image.png




case class Customers(custid:String,lname:String,fname:String,cardno:String,addr:String,area:String,city:String,language:String,score:String)
case class Orders(orderid:String,orddate:String,custid:String,status:String)
case class OrderItem(oiid:String,ordid:String,proid:String,buynum:String,sp:String,cp:String)
case class Products(proid:String,protype:String,proname:String,price:String,photo:String)
// 效率高 方法一
val custs = sc.textFile("hdfs://192.168.56.137:9000/20200106/customers.csv").map(str=>{
var e = str.replaceAll("\"","").split(",")
Customers(e(0),e(1),e(2),e(3),e(4),e(5),e(6),e(7),e(8))
})
scala> custs.toDS.show(1)
+------+-------+---------+---------+---------+------------------+-----------+--------+-----+
|custid|  lname|    fname|   cardno|     addr|              area|       city|language|score|
+------+-------+---------+---------+---------+------------------+-----------+--------+-----+
|     1|Richard|Hernandez|XXXXXXXXX|XXXXXXXXX|6303 Heather Plaza|Brownsville|      TX|78521|
+------+-------+---------+---------+---------+------------------+-----------+--------+-----+
// 方法二
val custs = sc.textFile("hdfs://192.168.56.137:9000/20200106/customers.csv").map(_.replaceAll("\"","").split(",")).map(x=>Customers(x(0),x(1),x(2),x(3),x(4),x(5),x(6),x(7),x(8))).toDS.show(1)
val ords = sc.textFile("hdfs://192.168.56.137:9000/20200106/orders.csv").map(_.replaceAll("\"","").split(",")).map(x=>Orders(x(0),x(1),x(2),x(3))).toDS
val items = sc.textFile("hdfs://192.168.56.137:9000/20200106/order_items.csv").map(_.replaceAll("\"","").split(",")).map(x=>OrderItem(x(0),x(1),x(2),x(3),x(4),x(5))).toDS


3、使用DataSet完成Demo


  • 如上我们已经使用RDD装载业务数据
  • 定义样例类
  • 将RDD转换为DataSet
// 连接查询  谁的消费额最高 三表查询
val tabitem = items.groupBy("ordid").agg(sum($"cp").as("countPrice"))
val tabord = tabitem.join(ords,tabitem("ordid")===ords("orderid"))
tabord.show(2)
+-----+----------+-------+-------------------+------+---------------+           
|ordid|countPrice|orderid|            orddate|custid|         status|
+-----+----------+-------+-------------------+------+---------------+
|10096|    369.97|  10096|2013-09-25 00:00:00| 10503|PENDING_PAYMENT|
|10351|    299.98|  10351|2013-09-27 00:00:00|   723|        ON_HOLD|
+-----+----------+-------+-------------------+------+---------------+
tabord.where("ordid=1").show()
+-----+----------+-------------------+------+------+                            
|ordid|countPrice|            orddate|custid|status|
+-----+----------+-------------------+------+------+
|    1|    299.98|2013-07-25 00:00:00| 11599|CLOSED|
+-----+----------+-------------------+------+------+
tabord.orderBy(desc("countPrice")).show(3)
+-----+------------------+-------------------+------+--------+                  
|ordid|        countPrice|            orddate|custid|  status|
+-----+------------------+-------------------+------+--------+
|68703|3089.9500000000003|2013-08-16 00:00:00|  9515|COMPLETE|
|68724|           2609.93|2013-09-26 00:00:00|  1148|COMPLETE|
|68809|           2579.94|2014-03-12 00:00:00|  5946| ON_HOLD|
+-----+------------------+-------------------+------+--------+
tabord.orderBy(desc("countPrice")).limit(1).show
+-----+------------------+-------------------+------+--------+                  
|ordid|        countPrice|            orddate|custid|  status|
+-----+------------------+-------------------+------+--------+
|68703|3089.9500000000003|2013-08-16 00:00:00|  9515|COMPLETE|
+-----+------------------+-------------------+------+--------+                                          
val tabod=tabord.orderBy(desc("countPrice")).limit(1)
tabod.join(custs,tabod("custid")===custs("custid"))
.select(col("lname"),col("fname"),col("countPrice")).show
+--------+-----+------------------+                                             
|   lname|fname|        countPrice|
+--------+-----+------------------+
|Victoria|Smith|3089.9500000000003|
+--------+-----+------------------+
// 哪个产品销量最高 
val procounts =items.groupBy("proid")
.agg(sum($"buynum").as("countnum"))
.orderBy(desc("countnum")).limit(1)
procounts
.join(pros,procounts("proid")===pros("proid"))
.drop(pros("proid"))
.select(col("proid"),col("protype"),col("proname"),col("countnum")).show
+-----+-------+--------------------+--------+                                   
|proid|protype|             proname|countnum|
+-----+-------+--------------------+--------+
|  365|     17|Perfect Fitness P...| 73698.0|
+-----+-------+--------------------+--------+


3.3　DataFrame


image.png


在Spark中，DataFrame是一种以RDD为基础的分布式数据集，类似于传统数据库中的二维表格。DataFrame与RDD的主要区别在于，前者带有schema元信息，即DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得Spark SQL得以洞察更多的结构信息，从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。反观RDD，由于无从得知所存数据元素的具体内部结构，Spark Core只能在stage层面进行简单、通用的流水线优化。


image.png


1、创建 DataFrame

scala> val cus = spark.read.format("csv").option("header","true").load("hdfs://192.168.56.137:9000/20200102/events.csv")
// 使用printSchema方法输出DataFrame的Schema信息
scala> cus.printSchema
// 表头有名字 直接注册一个临时表
scala> cus.registerTempTable("users")
// sql()方法执行SQL查询操作
scala> spark.sql("select * from users").show(2)
scala> spark.sql("select event_id,user_id from users").show(2)


// 表头没名字
scala> val cs = spark.read.format("csv").load("hdfs://192.168.56.137:9000/20200106/customers.csv")
scala> cs.printSchema
root
 |-- _c0: string (nullable = true)
 |-- _c1: string (nullable = true)
 |-- _c2: string (nullable = true)
 |-- _c3: string (nullable = true)
 |-- _c4: string (nullable = true)
 |-- _c5: string (nullable = true)
 |-- _c6: string (nullable = true)
 |-- _c7: string (nullable = true)
 |-- _c8: string (nullable = true)
// 单个命名
scala> cs.withColumnRenamed("_c0","id")
// 多个命名
scala> val lookup = Map("_c0"->"id","_c1"->"lname")
scala> cs.select(cs.columns.map(c=>col(c).as(lookup.getOrElse(c,c))):_*)
scala> cs.select(cs.columns.map(c=>col(c).as(lookup.getOrElse(c,c))):_*).registerTempTable("abc")
scala> spark.sql("select id,lname from abc").show(2)
+---+-------+
| id|  lname|
+---+-------+
|  1|Richard|
|  2|   Mary|
+---+-------+


2、使用DataFrame完成Demo

val custs = spark.read.format("csv").load("hdfs://192.168.56.137:9000/20200106/customers.csv")
val ords = spark.read.format("csv").load("hdfs://192.168.56.137:9000/20200106/orders.csv")
val items = spark.read.format("csv").load("hdfs://192.168.56.137:9000/20200106/order_items.csv")
val pros = spark.read.format("csv").load("hdfs://192.168.56.137:9000/20200106/products.csv")
val lookup = Map("_c0"->"ordid","_c1"->"orddate","_c2"->"custid","_c3"->"status")
ords.select(ords.columns.map(c=>col(c).as(lookup.getOrElse(c,c))):_*).registerTempTable("ordstab")
spark.sql("select dayofweek(orddate) as week,count(ordid) as countnum from ordstab group by week")
+----+--------+                                                                 
|week|countnum|
+----+--------+
|   1|    9735|
|   6|   10288|
|   3|    9964|
|   5|    9862|
|   4|    9758|
|   7|    9984|
|   2|    9292|
+----+--------+
val lookup = Map("_c0"->"custid","_c1"->"lname","_c2"->"fname","_c3"->"cardno","_c4"->"addr","_c5"->"area","_c6"->"city","_c7"->"language","_c8"->"score")
custs.select(custs.columns.map(c=>col(c).as(lookup.getOrElse(c,c))):_*).registerTempTable("custstab")
val lookup = Map("_c0"->"oiid","_c1"->"ordid","_c2"->"proid","_c3"->"buynum","_c4"->"sp","_c5"->"cp")
items.select(items.columns.map(c=>col(c).as(lookup.getOrElse(c,c))):_*).registerTempTable("itemstab")
spark.sql("select c.city ,sum(t.sp) as total_purchasses from itemstab as t inner join ordstab as o on o.ordid=t.ordid inner join custstab as c on c.custid=o.custid group by c.city order by total_purchasses desc")


文章标签：
云解析DNS
分布式计算
SQL
关系型数据库
MySQL
Spark
HIVE
Scala
API
数据库
编译器
关键词：
云解析DNS应用
云解析DNS sql
apache spark SQL
sql云解析DNS
SQL解析
云祁
目录
相关文章
游客dng4gjyb342he
|
4天前
RS-485网络中的标准端接与交流电端接应用解析
RS-485，作为一种广泛应用的差分信号传输标准，因其传输距离远、抗干扰能力强、支持多点通讯等优点，在工业自动化、智能建筑、交通运输等领域得到了广泛应用。在构建RS-485网络时，端接技术扮演着至关重要的角色，它直接影响到网络的信号完整性、稳定性和通信质量。
游客dng4gjyb342he
18 2
1778908750453455
|
10天前
|
自然语言处理 并行计算 数据可视化
免费开源法律文档比对工具：技术解析与应用
这款免费开源的法律文档比对工具，利用先进的文本分析和自然语言处理技术，实现高效、精准的文档比对。核心功能包括文本差异检测、多格式支持、语义分析、批量处理及用户友好的可视化界面，广泛适用于法律行业的各类场景。
1778908750453455
32 1
一缕微风绕指柔
|
4天前
|
存储 供应链 物联网
深入解析区块链技术的核心原理与应用前景
深入解析区块链技术的核心原理与应用前景
一缕微风绕指柔
12 0
跃@sir
|
4天前
|
存储 供应链 安全
深度解析区块链技术的核心原理与应用前景
深度解析区块链技术的核心原理与应用前景
跃@sir
12 0
赵渝强老师
|
8天前
|
SQL JSON 分布式计算
【赵渝强老师】Spark SQL的数据模型：DataFrame
本文介绍了在Spark SQL中创建DataFrame的三种方法。首先，通过定义case class来创建表结构，然后将CSV文件读入RDD并关联Schema生成DataFrame。其次，使用StructType定义表结构，同样将CSV文件读入RDD并转换为Row对象后创建DataFrame。最后，直接加载带有格式的数据文件（如JSON），通过读取文件内容直接创建DataFrame。每种方法都包含详细的代码示例和解释。
赵渝强老师
25 0
陌陌谣
|
8天前
|
SQL 监控 安全
员工上网行为监控软件：SQL 在数据查询监控中的应用解析
在数字化办公环境中，员工上网行为监控软件对企业网络安全和管理至关重要。通过 SQL 查询和分析数据库中的数据，企业可以精准了解员工的上网行为，包括基础查询、复杂条件查询、数据统计与分析等，从而提高网络管理和安全防护的效率。
陌陌谣
21 0
lfreeali
|
SQL Java 关系型数据库
V$SQLAREA解析
V$SQLAREA lists statistics on shared SQL area and contains one row per SQL string.
lfreeali
817 0
游客j4mujezz7vm7y
|
2月前
|
关系型数据库 MySQL 网络安全
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
5-10Can't connect to MySQL server on 'sh-cynosl-grp-fcs50xoa.sql.tencentcdb.com' (110)")
游客j4mujezz7vm7y
36 0
sunrr
|
4月前
|
SQL 存储 监控
SQL Server的并行实施如何优化？
【7月更文挑战第23天】SQL Server的并行实施如何优化？
sunrr
111 13
小王老师呀
|
4月前
|
SQL
解锁 SQL Server 2022的时间序列数据功能
【7月更文挑战第14天】要解锁SQL Server 2022的时间序列数据功能，可使用`generate_series`函数生成整数序列，例如：`SELECT value FROM generate_series(1, 10)。此外，`date_bucket`函数能按指定间隔（如周）对日期时间值分组，这些工具结合窗口函数和其他时间日期函数，能高效处理和分析时间序列数据。更多信息请参考官方文档和技术资料。
小王老师呀
71 9

热门文章

最新文章

  • 1
    Core Image 框架详细解析
  • 2
    DNS害我不能上网,找几个备用
  • 3
    Windows登录类型及安全日志解析
  • 4
    golang开发:类库篇(四)配置文件解析器goconfig的使用
  • 5
    ARP解析MAC地址的全过程（ARP的工作机制）
  • 6
    《R语言数据挖掘：实用项目解析》——第2章，第2.9节无参数方法
  • 7
    信息安全之DNS欺骗详解
  • 8
    DNS服务
  • 9
    java代码注释指南：三大注释解析与最佳实践
  • 10
    dns,lamp,文件共享服务为基础的三个示例
  • 1
    DAG（有向无环图）在Spark中的应用
    427
  • 2
    scala-spark中的groupby、like等的用法
    220
  • 3
    spark执行sql的原理是什么
    124
  • 4
    容器服务Kubernetes版产品使用合集之怎么实现把 spark 跑在k8s
    87
  • 5
    spark用于分析数据并将数据保存到数据库中是
    70
  • 6
    数据分享|Python、Spark SQL、MapReduce决策树、回归对车祸发生率影响因素可视化分析
    88
  • 7
    如何解决Spark在深度学习中的资源消耗问题？
    88
  • 8
    Spark在深度学习中的优缺点是什么？
    233
  • 9
    使用Spark进行机器学习
    69
  • 10
    Spark是一个基于内存的通用数据处理引擎，可以进行大规模数据处理和分析
    132

    • 相关课程

    更多
  • SQL入门与实践
  • 数据库及SQL/MySQL基础
  • SQL进阶及查询
  • Java面试疑难点解析 - 面试技巧及语言基础
  • Java面试疑难点解析 - Java Web开发
  • Java面试疑难点解析 - 系统架构及项目设计

    • 相关电子书

    更多
  • SQL Server在电子商务中的应用与实践
  • GeoMesa on Spark SQL
  • 原生SQL on Hadoop引擎- Apache HAWQ 2.x最新技术解密malili

    • 相关实验场景

    更多
  • SQL的增删改查及函数应用
  • PolarDB MySQL自动SQL限流
  • 如何开通SQL Server的专属集群
  • 使用DAS实现数据库自动SQL优化
  • 使用DAS实现数据库自动SQL限流
  • 一小时快速掌握 SQL 语法

    • 推荐镜像

    更多
  • DNS
  • kali-security
  • NTP
    • 下一篇
    阿里云OSS设置跨域访问