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

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

2022-06-10 112
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
云解析 DNS，旗舰版 1个月
云解析DNS，个人版 1个月
全局流量管理 GTM，标准版 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
数据库
编译器
关键词：
sql云解析DNS
SQL解析
云解析DNS sql
云解析DNS应用
SQL应用
云祁
目录
相关文章
听风的鱼鱼儿
|
7天前
|
消息中间件 运维 监控
Linux命令lsipc：深入解析与实战应用
`lsipc` (通常指 `ipcs`) 是Linux命令，用于查看系统中的IPC资源，包括消息队列、信号量和共享内存。它显示详细信息，支持过滤，并且需要相应权限。示例用法：显示共享内存(`-m`)、查询消息队列(`-q -i ID`)、查看关联进程(`-m -p`)。注意权限、操作影响及定期监控。结合`ipcrm`等工具可进行更深入管理。
听风的鱼鱼儿
27 0
梦回故国楼台梦
|
3天前
|
前端开发 小程序 Java
深入解析Java Servlet与JSP：构建高效服务器端应用
【6月更文挑战第23天】Java Servlet和JSP是Web开发的关键技术，用于构建高效服务器端应用。Servlet处理HTTP请求，执行业务逻辑，而JSP专注于动态HTML生成。两者结合，借助MVC架构，实现逻辑与视图分离，提高代码可读性和性能。尽管有新框架出现，Servlet和JSP仍是许多项目的基础。
梦回故国楼台梦
18 7
做梦都在改BUG
|
2天前
|
数据采集 算法 BI
解析numpy中的iscomplex方法及实际应用
在 NumPy 中，iscomplex 函数用于检查数组中的每个元素是否为复数。这个函数在处理包含复数数据的数组时非常有用，尤其是在科学计算和工程领域，这些领域经常需要区分实数和复数。 在数学和工程领域，复数是一种基本的数值类型，它们扩展了实数系统，包含了实部和虚部。在 NumPy 中，复数由 numpy.complex128 或 numpy.complex64 类型表示。numpy.iscomplex 函数提供了一种简便的方式来检查数组中的元素是否为复数。这对于数据类型判断、数据清洗和后续的数值分析非常重要。
做梦都在改BUG
10 3
听风的鱼鱼儿
|
3天前
|
NoSQL Linux 程序员
Linux objdump命令：深入解析与实战应用
`objdump`是Linux下的反汇编工具，用于将二进制文件转换为汇编代码，便于理解程序底层。它可以反汇编目标文件、可执行文件和库，支持多种参数，如显示符号表(-t)、反汇编代码(-d)、源代码与汇编混合视图(-S)。在实践中，结合-g编译选项和特定段(-j)反汇编，能辅助调试和分析。使用时注意包含调试信息，选择适当参数，并与其他工具（如gdb）配合使用。
听风的鱼鱼儿
18 1
蓝易云
|
4天前
|
关系型数据库 MySQL 数据库连接
蓝易云 - PHP基本语法解析与应用指南
以上只是PHP基本语法的简要概述，要深入了解和掌握PHP，你需要阅读更多的教程和参考资料，并通过实践来提高你的技能。
蓝易云
18 2
郑小健
|
5天前
|
数据可视化 搜索推荐 atlas
DataV Atlas深度解析与实战应用：打造个性化地理信息可视化
阿里云DataV的Atlas功能专注于地理信息可视化，提供范围选择、边界生成和层级展示等工具，助用户轻松创建专业地图应用。通过代码示例展示了如何用Geo组件展示中国省份销售数据，强调了数据安全和性能优化的重要性。DataV Atlas简化了复杂地理信息的展示，提升了数据洞察的直观性和美感。【6月更文挑战第19天】
郑小健
380 3
米诺斯
|
1天前
|
SQL Java 数据库连接
Mybatis日志SQL解析
Mybatis日志SQL解析
米诺斯
5 0
richard_yuu
|
7天前
|
机器学习/深度学习 人工智能 自然语言处理
【数字人】AIGC技术引领数字人革命：从制作到应用到全景解析
【数字人】AIGC技术引领数字人革命：从制作到应用到全景解析
richard_yuu
21 0
半夜敲代码的夜猫子
|
8天前
|
SQL Java 数据库连接
JavaWeb基础第三章（MyBatis的应用，基础操作与动态SQL）
JavaWeb基础第三章（MyBatis的应用，基础操作与动态SQL）
半夜敲代码的夜猫子
14 0
zhanbao
|
8天前
|
运维 应用服务中间件 数据库
深入解析现代运维中的自动化工具应用
在现代运维领域，自动化工具成为提高工作效率和降低人为错误的关键因素。本文将探讨几种常见的运维自动化工具，它们的功能、优势及其在实际应用中的案例，以期为运维人员提供有价值的参考。
zhanbao
10 0

热门文章

最新文章

  • 1
    《DNS攻击防范科普系列4》--遭遇DNS缓存投毒该怎么办？
  • 2
    Unity 消息发送机制 解析
  • 3
    《深入解析Android 虚拟机》——第2章，第2.2节Java虚拟机概述
  • 4
    Linux命令行(Console)解析JSON
  • 5
    C语言文件操作解析(五)之EOF解析
  • 6
    利用GSON解析JSON数据
  • 7
    Android事件分发机制完全解析，带你从源码的角度彻底理解
  • 8
    Android系统原生应用解析之桌面闹钟及相关原理应用之时钟任务的应用(二)
  • 9
    fragment和fragmentactivity解析
  • 10
    使用awk来解析dump文件
  • 1
    Spark学习---SparkSQL（概述、编程、数据的加载和保存、自定义UDFA、项目实战）
    187
  • 2
    安装Spark
    49
  • 3
    Spark学习---day07、Spark内核（Shuffle、任务执行）
    65
  • 4
    Spark读取变更Hudi数据集Schema实现分析
    78
  • 5
    实战|使用Spark Streaming写入Hudi
    119
  • 6
    adb spark的lakehouse api访问内表数据，还支持算子下推吗
    115
  • 7
    Spark学习---day06、Spark内核（源码提交流程、任务执行）
    62
  • 8
    Spark学习--day05、SparkCore电商网站实操、SparkCore-工程代码
    83
  • 9
    Spark【Spark学习大纲】简介+生态+RDD+安装+使用（xmind分享）
    47
  • 10
    Spark【基础知识 03】【RDD常用算子详解】（图片来源于网络）
    73

    • 相关课程

    更多
  • 第八届大学生创新创业大赛阿里命题IoT赛题解析
  • 云计算工程师解析与实战-网络专家篇（体验版）
  • 深入解析Docker容器化技术
  • Java面试疑难点解析 - 面试技巧及语言基础
  • Java面试疑难点解析 - Java Web开发
  • Java面试疑难点解析 - 系统架构及项目设计

    • 相关电子书

    更多
  • 神龙云服务器产品及技术深度解析
  • 弹性创造价值：基于ECS的最佳性价比实践解析
  • 又快又稳：阿里云下一代虚拟交换机解析

    • 相关实验场景

    更多
  • PolarDB for AI：在数据库中通过SQL实现AI能力
  • 玩转MaxCompute SQL! 30分钟搞定数据分析挖掘
  • 通过云拨测对指定服务器进行Ping/DNS监测
  • 使用SQL语句实现数据插入、修改和删除操作
  • 使用SQL语句实现数据查询操作
  • 使用SQL语句管理索引

    • 推荐镜像

    更多
  • DNS
  • NTP
  • kali-security
    • 下一篇
    部署LAMP环境（Alibaba Cloud Linux 3）