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

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

2022-06-10 122
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云解析 DNS，旗舰版 1个月
全局流量管理 GTM，标准版 1个月
公共DNS（含HTTPDNS解析），每月1000万次HTTP解析
简介： 【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应用
sql云解析DNS
SQL解析
apache spark应用
SQL应用
云祁
目录
相关文章
众所周知
|
3天前
|
存储 算法 Java
解析HashSet的工作原理，揭示Set如何利用哈希算法和equals()方法确保元素唯一性，并通过示例代码展示了其“无重复”特性的具体应用
在Java中，Set接口以其独特的“无重复”特性脱颖而出。本文通过解析HashSet的工作原理，揭示Set如何利用哈希算法和equals()方法确保元素唯一性，并通过示例代码展示了其“无重复”特性的具体应用。
众所周知
15 3
游客cyvjzkyhhhmiw
|
5天前
|
开发框架 供应链 监控
并行开发模型详解：类型、步骤及其应用解析
在现代研发环境中，企业需要在有限时间内推出高质量的产品，以满足客户不断变化的需求。传统的线性开发模式往往拖慢进度，导致资源浪费和延迟交付。并行开发模型通过允许多个开发阶段同时进行，极大提高了产品开发的效率和响应能力。本文将深入解析并行开发模型，涵盖其类型、步骤及如何通过辅助工具优化团队协作和管理工作流。
游客cyvjzkyhhhmiw
30 3
游客5fdji2pvmf8888
|
7天前
|
前端开发
深入解析React Hooks：构建高效且可维护的前端应用
本文将带你走进React Hooks的世界，探索这一革新特性如何改变我们构建React组件的方式。通过分析Hooks的核心概念、使用方法和最佳实践，文章旨在帮助你充分利用Hooks来提高开发效率，编写更简洁、更可维护的前端代码。我们将通过实际代码示例，深入了解useState、useEffect等常用Hooks的内部工作原理，并探讨如何自定义Hooks以复用逻辑。
游客5fdji2pvmf8888
25 1
叫个什么名字
|
6天前
|
设计模式 PHP 开发者
PHP中的设计模式：桥接模式的解析与应用
在软件开发的浩瀚海洋中，设计模式如同灯塔一般，为开发者们指引方向。本文将深入探讨PHP中的一种重要设计模式——桥接模式。桥接模式巧妙地将抽象与实现分离，通过封装一个抽象的接口，使得实现和抽象可以独立变化。本文将阐述桥接模式的定义、结构、优缺点及其应用场景，并通过具体的PHP示例代码展示如何在实际项目中灵活运用这一设计模式。让我们一起走进桥接模式的世界，感受它的魅力所在。
叫个什么名字
16 5
1100237741946300
|
4天前
|
SQL 监控 数据库
SQL语句是否都需要解析及其相关技巧和方法
在数据库管理中，SQL（结构化查询语言）语句的使用无处不在，它们负责数据的查询、插入、更新和删除等操作
1100237741946300
12 1
喜欢猪猪
|
6天前
|
架构师 关系型数据库 MySQL
MySQL最左前缀优化原则：深入解析与实战应用
【10月更文挑战第12天】在数据库架构设计与优化中，索引的使用是提升查询性能的关键手段之一。其中，MySQL的最左前缀优化原则（Leftmost Prefix Principle）是复合索引（Composite Index）应用中的核心策略。作为资深架构师，深入理解并掌握这一原则，对于平衡数据库性能与维护成本至关重要。本文将详细解读最左前缀优化原则的功能特点、业务场景、优缺点、底层原理，并通过Java示例展示其实现方式。
喜欢猪猪
16 1
土木林森
|
7天前
|
机器学习/深度学习 人工智能 算法
揭开深度学习与传统机器学习的神秘面纱：从理论差异到实战代码详解两者间的选择与应用策略全面解析
【10月更文挑战第10天】本文探讨了深度学习与传统机器学习的区别，通过图像识别和语音处理等领域的应用案例，展示了深度学习在自动特征学习和处理大规模数据方面的优势。文中还提供了一个Python代码示例，使用TensorFlow构建多层感知器（MLP）并与Scikit-learn中的逻辑回归模型进行对比，进一步说明了两者的不同特点。
土木林森
26 2
1100237741946300
|
4天前
|
SQL 数据可视化 BI
SQL语句及查询结果解析：技巧与方法
在数据库管理和数据分析中，SQL语句扮演着至关重要的角色
1100237741946300
32 0
喜欢猪猪
|
5天前
|
XML Java 数据格式
Spring IOC容器的深度解析及实战应用
【10月更文挑战第14天】在软件工程中，随着系统规模的扩大，对象间的依赖关系变得越来越复杂，这导致了系统的高耦合度，增加了开发和维护的难度。为解决这一问题，Michael Mattson在1996年提出了IOC（Inversion of Control，控制反转）理论，旨在降低对象间的耦合度，提高系统的灵活性和可维护性。Spring框架正是基于这一理论，通过IOC容器实现了对象间的依赖注入和生命周期管理。
喜欢猪猪
17 0
一位隐者
|
5天前
|
供应链 网络协议 数据安全/隐私保护
互联网络、互联网、内部网与外部网：概念解析与应用差异
【10月更文挑战第11天】
一位隐者
69 0

热门文章

最新文章

  • 1
    Spark-ML-01-小试spark分析离线商品信息
  • 2
    大数据平台搭建（容器环境）——Spark3.X on Yarn安装配置
  • 3
    Spark-神奇的共享变量
  • 4
    [Spark][Python]Mapping Single Rows to Multiple Pairs
  • 5
    工作经验分享：Spark调优【优化后性能提升1200%】
  • 6
    [Spark][Python]Spark Python 索引页
  • 7
    Spark Job的提交与task本地化分析（源码阅读八）
  • 8
    Spark-构建基于Spark的推荐引擎
  • 9
    spark 3.1.x支持（兼容）hive 1.2.x以及hadoop cdh版本的尝试
  • 10
    Spark-spark streaming
  • 1
    Elasticsearch 悬挂索引解析与管理指南
    77
  • 2
    Elasticsearch 通过索引阻塞实现数据保护深入解析
    60
  • 3
    日志解析神器——Logstash中的Grok过滤器使用详解
    477
  • 4
    Android12 双击power键启动相机源码解析
    167
  • 5
    API在电子商务中的应用与优势：深入解析
    64
  • 6
    Java8 Lambda实现源码解析
    162714
  • 7
    LISA微调技术解析：比LoRA更低的显存更快的速度
    389
  • 8
    深入解析JavaScript中的模板字符串
    70
  • 9
    linux centos系统搭建samba文件服务器 NetBIOS解析 （超详细）
    199
  • 10
    Linux中搭建DNS 域名解析服务器（详细版）
    345

    • 相关课程

    更多
  • 大数据实战项目 - 反爬虫系统（Lua+Spark+Redis+Hadoop框架搭建）第六阶段
  • SQL完全自学手册
  • SQL Server on Linux入门教程
  • SQL入门与实践
  • 大数据实时计算框架Spark快速入门
  • SQL进阶及查询

    • 相关电子书

    更多
  • Hybrid Cloud and Apache Spark
  • Scalable Deep Learning on Spark
  • Comparison of Spark SQL with Hive

    • 相关实验场景

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

    • 推荐镜像

    更多
  • DNS
  • NTP
  • wireshark
    • 下一篇
    AI助理直击要害，从繁复中提炼精华——使用CDN加速访问OSS存储的图片