Spark SQL 概述

Spark SQL 概述

Spark SQL 是 Apache Spark 的一个模块,专门用于处理结构化数据。它集成了 SQL 查询和 Spark 编程的强大功能,使得处理大数据变得更加高效和简便。通过 Spark SQL,用户可以直接在 Spark 中使用 SQL 查询,或者使用 DataFrame 和 DataSet API 进行数据操作。

  • 一、Spark SQL 架构
  • 二、Spark SQL 特点
  • 三、Spark SQL 运行原理
  • 四、Spark SQL API 相关概述
  • 五、Spark SQL 依赖
  • 六、Spark SQL 数据集
    • 1、DataFrame
    • 2、Dataset
    • 3、DataFrame 和 Dataset 的关系
  • 七、Spark Sql 基本用法
    • 1、Scala 创建 SparkSession 对象
    • 2、DataFrame 和 Dataset 的创建方式
    • 3、DataFrame API

一、Spark SQL 架构

Spark SQL 的架构主要由以下几个组件组成:

  1. SparkSession:Spark 应用的统一入口点,用于创建 DataFrame、DataSet 和执行 SQL 查询。
  2. Catalyst 优化器:Spark SQL 的查询优化引擎,负责解析、分析、优化和生成物理执行计划。
  3. DataFrame 和 DataSet API:提供面向对象的编程接口,支持丰富的数据操作方法。
  4. 数据源接口:支持多种数据源,如 HDFS、S3、HBase、Cassandra、Hive 等。
  5. 执行引擎:将优化后的查询计划转换为执行任务,并在分布式集群上并行执行这些任务。

二、Spark SQL 特点

  • 统一数据访问接口:支持多种数据源(如 CSV、JSON、Parquet、Hive、JDBC、HBase 等)并提供一致的查询接口。
  • DataFrame 和 Dataset API:提供面向对象的编程接口,支持类型安全的操作,便于数据处理。
  • Catalyst 优化器:自动将用户的查询转换为高效的执行计划,提升查询性能。
  • 与 Hive 的集成:无缝集成 Hive,能够直接访问现存的 Hive 数据,并使用 Hive 的 UDF 和 UDAF。
  • 高性能:通过 Catalyst 优化器和 Tungsten 执行引擎,实现高效的查询性能和内存管理。
  • 多种操作方式:支持 SQL 和 API 编程两种操作方式,灵活性高。
  • 外部工具接口:提供 JDBC/ODBC 接口供第三方工具借助 Spark 进行数据处理。
  • 高级接口:提供了更高层级的接口,方便地处理数据。

三、Spark SQL 运行原理

在这里插入图片描述

查询解析(Query Parsing):将 SQL 查询解析成抽象语法树(AST)。

逻辑计划生成(Logical Plan Generation):将 AST 转换为未优化的逻辑计划。

逻辑计划优化(Logical Plan Optimization):使用 Catalyst 优化器对逻辑计划进行一系列规则优化。

物理计划生成(Physical Plan Generation):将优化后的逻辑计划转换为一个或多个物理计划,并选择最优的物理计划。

执行(Execution):将物理计划转换为 RDD,并在集群上并行执行。

四、Spark SQL API 相关概述

SparkContext:SparkContext 是 Spark 应用程序的主入口点,负责连接到 Spark 集群,管理资源和任务调度。在 Spark 2.0 之后,推荐使用 SparkSession 取代 SparkContext。

SQLContext:SQLContext 是 Spark SQL 的编程入口点,允许用户通过 SQL 查询或 DataFrame API 进行数据处理。它提供了基本的 Spark SQL 功能。

HiveContext:HiveContext 是 SQLContext 的子集,增加了对 Hive 的集成支持,可以直接访问 Hive 中的数据和元数据,使用 Hive 的 UDF 和 UDAF。

SparkSession:SparkSession 是 Spark 2.0 引入的新概念,合并了 SQLContext 和 HiveContext 的功能,提供了统一的编程接口。SparkSession 是 Spark SQL 的建议入口点,支持使用 DataFrame 和 Dataset API 进行数据处理。

创建 SparkContext 和 SparkSession 的注意事项:如果同时需要创建 SparkContext 和 SparkSession,必须先创建 SparkContext,再创建 SparkSession。如果先创建 SparkSession,再创建 SparkContext,会导致异常,因为在同一个 JVM 中只能运行一个 SparkContext。

五、Spark SQL 依赖

<properties><spark.version>3.1.2</spark.version><spark.scala.version>2.12</spark.scala.version>
</properties>
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_${spark.scala.version}</artifactId><version>${spark.version}</version>
</dependency>

六、Spark SQL 数据集

在 Spark SQL 中,数据集主要分为以下几种类型:DataFrame 和 Dataset。它们是处理和操作结构化和半结构化数据的核心抽象。

1、DataFrame

Dataset 是在 Spark 2.0 中引入的新的抽象数据结构,它是强类型的,可以存储 JVM 对象。Dataset API 结合了 DataFrame 的操作简便性和类型安全性,适用于需要更高级别数据类型控制和面向对象编程风格的场景。具体特点如下:

  • 类似于二维表格:DataFrame 类似于传统的关系数据库中的二维表格。
  • Schema(数据结构信息):在 RDD 的基础上加入了 Schema,描述数据结构的信息。
  • 支持嵌套数据类型:DataFrame 的 Schema 支持嵌套的数据类型,如 structmaparray
  • 丰富的 SQL 操作 API:提供更多类似 SQL 操作的 API,便于进行数据查询和操作。

2、Dataset

Dataset 是在 Spark 2.0 中引入的新的抽象数据结构,它是强类型的,可以存储 JVM 对象。Dataset API 结合了 DataFrame 的操作简便性和类型安全性,适用于需要更高级别数据类型控制和面向对象编程风格的场景。具体特点如下:

  • 强类型:Spark 1.6中引入的一个更通用的数据集合,Dataset 是强类型的,提供类型安全的操作。
  • RDD + Schema:可以认为 Dataset 是 RDD 和 Schema 的结合,既有 RDD 的分布式计算能力,又有 Schema 描述数据结构的信息。
  • 适用于特定领域对象:可以存储和操作特定领域对象的强类型集合。
  • 并行操作:可以使用函数或者相关操作并行地进行转换和操作。

3、DataFrame 和 Dataset 的关系

  • DataFrame 是特殊的 Dataset:DataFrame 是 Dataset 的一个特例,即 DataFrame = Dataset[Row]
  • 数据抽象和操作方式的统一:DataFrame 和 Dataset 统一了 Spark SQL 的数据抽象和操作方式,提供了灵活且强大的数据处理能力。

七、Spark Sql 基本用法

1、Scala 创建 SparkSession 对象

import org.apache.spark.sql.SparkSession
object SparkSqlContext {def main(args: Array[String]): Unit = {// 创建 SparkConf 对象,设置应用程序的配置val conf: SparkConf = new SparkConf().setMaster("local[4]")   // 设置本地运行模式,使用 4 个线程.setAppName("spark sql") // 设置应用程序名称为 "spark sql"// 创建 SparkSession 对象,用于 Spark SQL 的编程入口val spark: SparkSession = SparkSession.builder().config(conf) // 将 SparkConf 配置应用于 SparkSession.getOrCreate() // 获取现有的 SparkSession,或者新建一个// 获取 SparkContext 对象,可以直接从 SparkSession 中获取val sc: SparkContext = spark.sparkContext// 导入 SparkSession 的隐式转换,可以使用 DataFrame API 的方法import spark.implicits._// 在这里可以编写数据处理代码,例如创建 DataFrame 和 Dataset,进行数据操作等...// 停止 SparkSession,释放资源spark.stop()}
}

2、DataFrame 和 Dataset 的创建方式

1、从集合创建

case class Person(name: String, age: Int)				// 下同val data1 = Seq(Person("Alice", 25), Person("Bob", 30))	
val ds: Dataset[Person] = spark.createDataset(data)		// 这里的spark是SparkSession对象(如上代码),下同val data2 = Seq(("Alice", 25), ("Bob", 30))
val df: DataFrame = data.toDF("name", "age")

1、从文件系统读取

val schema = StructType(Seq(StructField("name", StringType, nullable = false),StructField("age", IntegerType, nullable = false)
))val dsJson: Dataset[Person] = spark.read.json("/path/to/json/file").as[Person]val dfCsv: DataFrame = spark.read// 使用.schema方法指定CSV文件的模式(schema)其定义了DataFrame的列名和类型。// 这是一个可选步骤,但如果CSV文件没有头部行,或者你想覆盖文件中的头部行,则必须指定。  .schema(schema)		  // 这里设置"header"为"true",表示CSV文件的第一行是列名,不需要Spark从文件中自动推断。 .option("header", "true").csv("/path/to/csv/file")

3、从关系型数据库读取

val url = "jdbc:mysql://localhost:3306/database"
val properties = new java.util.Properties()
properties.setProperty("user", "username")
properties.setProperty("password", "password")val dsDb: Dataset[Person] = spark.read.jdbc(url, "table", properties).as[Person]val dfDb: DataFrame = spark.read.jdbc(url, "table", properties)

4、从非结构化数据源读取

val dsParquet: Dataset[Person] = spark.read.parquet("/path/to/parquet/file").as[Person]val dfParquet: DataFrame = spark.read.parquet("/path/to/parquet/file")

5、手动创建 Dataset

import org.apache.spark.sql.types._val schema = StructType(Seq(StructField("name", StringType, nullable = false),StructField("age", IntegerType, nullable = false)
))
val data = Seq(Row("Alice", 25), Row("Bob", 30))val dsManual: Dataset[Person] = spark.createDataFrame(spark.sparkContext.parallelize(data), schema).as[Person]val dfManual: DataFrame = spark.createDataFrame(spark.sparkContext.parallelize(data), schema
)

3、DataFrame API

语法示例一

模拟数据(1000条):

id,name,gender,age,city
1,邵睿,男,12,上海市
2,林子异,男,48,广州市
3,孟秀英,女,46,上海市
4,金嘉伦,男,8,北京市
...

需求:哪些城市和性别组合在人口较多(ID数量>50)的情况下具有最高的平均年龄,以及这些组合在各自性别中的排名。

// 导入SparkSession的隐式转换,这样可以使用DataFrame的便捷方法(例如下面的'$'符号)
import spark.implicits._// 定义了一个DataFrame的schema,但在这个例子中,使用了CSV的header来自动推断schema
val schema = StructType(Seq(StructField("id", LongType),StructField("name", StringType),StructField("gender", StringType),StructField("age", IntegerType),StructField("city", StringType),
))// 定义WindowSpec,用于后续的窗口函数操作,按gender分区,按avg_age降序排序,(复用使用此)
val WindowSpec: WindowSpec = Window.partitionBy($"gender").orderBy($"avg_age".desc)// 从CSV文件中读取数据,使用header作为列名,然后选择特定的列,进行分组和聚合操作
// 哪些城市和性别组合在人口较多(ID数量>50)的情况下具有最高的平均年龄,以及这些组合在各自性别中的排名。
spark.read// .schema(schema)	// 应用我们定义的schema .option("header", "true") 							// 使用CSV的header作为列名.csv("D:\\projects\\sparkSql\\people.csv")			// DataFrame.select($"id", $"name", $"age", $"city", $"gender") 	// 选择需要的列(不写默认就是全选).groupBy($"city", $"gender") 							// 按城市和性别分组.agg(			// 多重聚合count($"id").as("count"),   		// 计算每个组的ID数量round(avg($"age"), 2).as("avg_age") // 计算每个组的平均年龄,并保留两位小数).where($"count".gt(50))  		// 过滤出ID数量大于(可以使用>)50的组.orderBy($"avg_age".desc)     // 按平均年龄降序排序.select($"city", $"gender", $"avg_age",dense_rank().over(Window.partitionBy($"gender").orderBy($"avg_age".desc)).as("gender_avg_age_rank")).show() // 显示结果

结果:

+------+------+-------+-------------------+
|  city|gender|avg_age|gender_avg_age_rank|
+------+------+-------+-------------------+
|北京市|    男|  41.05|                  1|
|  东莞|    男|  42.81|                  2|
|上海市|    男|  43.92|                  3|
|成都市|    男|  45.89|                  4|
|  中山|    男|  47.08|                  5|
|广州市|    男|  47.47|                  6|
|  深圳|    男|  48.36|                  7|
|上海市|    女|  46.02|                  1|
|  中山|    女|  49.55|                  2|
+------+------+-------+-------------------+

语法示例二:视图,sql

// 读取CSV文件到DataFrame,使用header作为列名
val dfPeople: DataFrame = spark.read.option("header", "true") // 使用CSV的header作为列名.csv("D:\\projects\\sparkSql\\people.csv")// 将DataFrame注册为临时视图
dfPeople.createOrReplaceTempView("people_view")
// 可以使用Spark SQL来查询这个视图了
// 例如,查询所有人的姓名和年龄
spark.sql("SELECT name, age FROM people_view").show()
// 二
spark.sql("""|select * from people_view|where gender = '男'|""".stripMargin).show()

语法示例三:join

case class Student(name: String, classId: Int)
case class Class(classId: Int, className: String)val frmStu = spark.createDataFrame(Seq(Student("张三", 1),Student("李四", 1),Student("王五", 2),Student("赵六", 2),Student("李明", 2),Student("王刚", 4),Student("王朋", 5),)
)val frmClass = spark.createDataFrame(Seq(Class(1, "name1"),Class(2, "name2"),Class(3, "name3"),Class(4, "name4"))
)

left 左连接,rignt 右连接, full 全外连接,anti左差集,semi左交集

// 别名 + inner 内连接
frmStu.as("S").join(frmClass.as("C"), $"S.classId" === $"C.classId")	// joinType 默认 inner内连接.show()// 使用左外连接将df和frmClass根据classId合并
frmStu.join(frmClass, Seq("classId"), "left")	.show()// 左差集
frmStu.join(frmClass, Seq("classId"), "anti")	.show()// 左交集
frmStu.join(frmClass, Seq("classId"), "semi")	.show()

结果

别名 + inner 内连接
+----+-------+-------+---------+
|name|classId|classId|className|
+----+-------+-------+---------+
|张三|      1|      1|    name1|
|李四|      1|      1|    name1|
|王五|      2|      2|    name2|
|赵六|      2|      2|    name2|
|李明|      2|      2|    name2|
|王刚|      4|      4|    name4|
+----+-------+-------+---------+使用左外连接将df和frmClass根据classId合并
+-------+----+---------+
|classId|name|className|
+-------+----+---------+
|      1|张三|    name1|
|      1|李四|    name1|
|      2|王五|    name2|
|      2|赵六|    name2|
|      2|李明|    name2|
|      4|王刚|    name4|
|      5|王朋|     null|
+-------+----+---------+左差集
+-------+----+
|classId|name|
+-------+----+
|      5|王朋|
+-------+----+左交集
+-------+----+
|classId|name|
+-------+----+
|      1|张三|
|      1|李四|
|      2|王五|
|      2|赵六|
|      2|李明|
|      4|王刚|
+-------+----+

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/377753.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

JMeter进行HTTP接口测试的技术要点

参数化 用户定义的变量 用的时候 ${名字} 用户参数 在参数列表中传递 并且也是${} csv数据文件设置 false 不忽略首行 要首行 从第一行读取 true 忽略首行 从第二行开始 请求时的参数设置&#xff1a; 这里的名称是看其接口需要的请求参数的名称 这里的变量名称就是为csv里面…

itextpdf字体选择

itextpdf 版本7.2.5 itextpdf-html2pdf 版本4.0.5 这里讲的是通过html转pdf&#xff0c;在html2pdf中是通过html中font-family样式来确定字体的&#xff0c;那已知font-family的情况&#xff0c;怎么确定pdf中实际用的字体&#xff0c;大致分为两步&#xff1a; 1、通过font…

网页数据抓取:融合BeautifulSoup和Scrapy的高级爬虫技术

网页数据抓取&#xff1a;融合BeautifulSoup和Scrapy的高级爬虫技术 在当今的大数据时代&#xff0c;网络爬虫技术已经成为获取信息的重要手段之一。Python凭借其强大的库支持&#xff0c;成为了进行网页数据抓取的首选语言。在众多的爬虫库中&#xff0c;BeautifulSoup和Scrap…

图像识别和目标检测在超市电子秤上的应用

目录 前言深度学习的目标检测图像识别技术视觉秤的优势其他应用场景中的技术应用未来展望 前言 随着科技的不断发展&#xff0c;电子秤在生鲜超市中的应用也在不断升级。传统的电子秤需要打秤人员手动输入秤码&#xff0c;这不仅耗时费力&#xff0c;还需要大量的培训以记住各…

在Mac上一键安装Mysql(解决所有安装问题)

重点强调安装mysql成功的关键在于安装的版本不能是最新&#xff01;&#xff01; 目录 一&#xff1a;下载mysql数据库安装部分到此结束 二&#xff1a;配置mysql数据库三&#xff1a;启动mysql数据库四&#xff1a;各类奇葩问题总结 一&#xff1a;下载mysql数据库 1.进入MyS…

替换:show-overflow-tooltip=“true“ ,使用插槽tooltip,达到内容可复制

原生的show-overflow-tooltip“true” 不能满足条件&#xff0c;使用插槽自定义编辑&#xff1b; 旧code <el-table-column prop"reason" label"原因" align"center" :show-overflow-tooltip"true" /> <el-table-column pro…

SpringBoot系列:通过AOP+注解优雅实现操作日志记录

文章目录 前言一、简介1.1 操作日志在企业应用中的重要性1.2 使用AOP和注解实现操作日志记录的好处 二、开发环境三、准备工作3.1 创建操作日志记录表3.2 创建系统日志实体类 四、代码实现4.1 创建业务枚举类4.2 创建日志注解4.3 创建操作状态枚举类4.4 创建IP工具类4.5 创建切…

[A-04] ARMv8/ARMv9-Cache的相关策略

ver0.3 前言 前面我们已经通过三篇文章反反复复的讲Cache的概念、结构、架构&#xff0c;相信大家对Cache已经大概有了初步的了解。这里简单归纳一下: (1) Cache从硬件视角看&#xff0c;是连接PE-Core和主存的一种存储介质&#xff0c;存储的数据是主存中数据的副本&#xf…

【算法消费者行为】算法性能预期如何增强冲动性购买?—推文分享—2024-07-16

今天的推文主题是&#xff1a;【算法&消费者行为】 第一篇&#xff1a;如何通过管理多种刺激来优化客户体验&#xff1f;购买行为的驱动因素是隐性还是显性的消费者态度&#xff1f;第二篇&#xff1a;算法性能期望如何增强在线零售中的即兴购买行为。第三篇&#xff1a;社…

【Linux】Ubuntu 漏洞扫描与修复的吃瘪经历

自从上次“劫持”事情后&#xff0c;项目经理将所有跟安全相关的都推给我了&#xff08;不算 KPI 又要被白嫖&#xff0c;烦死了&#xff09;。这次客户又提了一个服务器安全扫描和漏洞修复的“活”&#xff0c;我这边顺手将过程记录一下&#xff0c;就当经验总结跟各位分享一下…

Parallels Desktop 19 for Mac(PD19虚拟机)详细图文安装教程分享

Parallels Desktop 19是一款功能丰富、性能强大且易于使用的虚拟机软件&#xff0c;它可以让您在Mac上同时运行多个操作系统&#xff0c;为您提供更大的灵活性和兼容性。 Parallels Desktop 19 for Mac(PD19虚拟机)下载安装包 Parallels Desktop 19 for Mac(PD19虚拟机)详细图…

【Visual Studio】Visual Studio使用技巧及报错解决合集

目录 目录 一.概述 二.Visual Studio报错问题及解决方法 三.Visual Studio操作过程中遇到的问题及解决方法 四.Visual Studio编译优化选项 五.Visual Studio快捷键 一.概述 持续更新Visual Studio报错及解决方法&#xff0c;包括Visual Studio报错问题及解决方法、Visua…

phenocycler(原CODEX)鉴定三阴乳腺癌不同的治疗反应轨迹

目前临床上需要制定策略来更好地识别哪些患者可以从单独的免疫疗法中受益&#xff0c;或者哪些患者可能需要化疗或放疗等额外疗法来克服耐药性。尽管一些联合放射治疗&#xff08;RT&#xff09;和免疫检查点抑制&#xff08;ICI&#xff09;的临床研究取得了成功&#xff0c;但…

【Linux】常见指令(下)

【Linux】常见指令&#xff08;下&#xff09; 通配符 *man指令cp指令echo指令cat指令&#xff08;简单介绍&#xff09;cp指令 mv指令alias指令which ctrl ccat指令linux下一切皆文件 more指令less指令head指令tail指令管道 通配符 ‘*’ 通配符’ *‘&#xff0c;是可以匹配…

怎样去除视频上的水印和文字,视频水印文本移除教程

在观看和分享视频时&#xff0c;我们经常会遇到带有水印或额外文字的情况。这些标记有时是为了版权保护&#xff0c;有时则是平台的标识&#xff0c;但在某些情况下&#xff0c;它们可能会干扰视频的观赏体验。本文将向你介绍常见的视频水印类型以及如何使用简鹿水印助手去除这…

ARM功耗管理之功耗数据与功耗收益评估

安全之安全(security)博客目录导读 思考&#xff1a;功耗数据如何测试&#xff1f;功耗曲线&#xff1f;功耗收益评估&#xff1f; UPF的全称是Unified Power Format&#xff0c;其作用是把功耗设计意图&#xff08;power intent&#xff09;传递给EDA工具&#xff0c; 从而帮…

05 以物品与用户为基础个性化推荐算法的四大策略

《易经》&#xff1a;“九二&#xff1a;见龙在田&#xff0c;利见大人”。九二是指阳爻在卦中处于第二位&#xff0c;见龙指龙出现在地面上&#xff0c;开始崭露头角&#xff0c;但是仍须努力&#xff0c;应处于安于偏下的位置。 本节是模块二第一节&#xff0c;模块二讲解传…

[Linux+git+Gitee+Jenkins]持续集成实验安装配置详细

首先理解持续集成原理&#xff0c;看懂并理解图 1。 图 1 持续集成原理结构 图 1 中&#xff0c;版本控制服务器指远程代码仓库&#xff0c;本实验使用 GitEE 作为远程代码仓库&#xff1b;Jenkins 自动化部署服务器为虚拟机&#xff0c;操作系统为 Linux &#xff1b…

人工智能与伦理挑战:多维度应对策略

人工智能技术近年来取得了迅猛发展&#xff0c;广泛应用于医疗诊断、金融分析、教育辅助、自动驾驶等各个领域&#xff0c;极大地提升了生产效率和服务质量&#xff0c;推动了科技进步和商业创新。然而&#xff0c;伴随其普及和应用的泛滥&#xff0c;AI也带来了数据隐私侵犯、…

C语言 | Leecode C语言题解之第229题多数元素II

题目&#xff1a; 题解&#xff1a; /*** Note: The returned array must be malloced, assume caller calls free().*//*假定 num1&#xff0c;num2 为出现次数大于 nums.length / 3 的两个数。&#xff08;最多出现两个&#xff09;遍历 nums&#xff0c; 若出现 num1、num2…