SparkSQL学习01

目录

    • 1.SparkSQL特点
      • 1.1易整合
      • 1.2统一的数据访问
      • 1.3兼容Hive
      • 1.4标准的数据连接
    • 2 SparkSQL编程模型=DataFrame=DataSet
      • 2.1 SQL
      • 2.2 DataFrame是什么
      • 2.3 DataSet是什么
      • 2.4 RDD,DataSet,DataFrame
    • 3 SparkSQL核心编程
      • 3.1 编程入口
      • 3.2 SparkSQL基本编程
        • 3.2.1编程入口SparkSession
        • 3.2.2 DSL语法 -->结合SQL中关键字作为函数(算子)的名字传递参数进行编程方式-->接近于RDD编程
        • 3.2.3 导入SparkSession中隐式转换操作,增强sql功能
        • 3.2.4 SQL语法 -->直接写SQL或者HQL语言进行编程【算是SparkSQL主流】

SparkSQL是Spark用于结构化数据处理的Spark模块,是Spark生态体系中的构建在SparkCore基础之上的一个基于SQL的计算模块,不依赖于Hive。

SparkSQL与基本的SparkRDDAPI不同,SparksQL提供的接口为Spark提供了有关数据结构和正在执行的计算的更多信息。在内部,SparkSQL使用这些额外的信息来执行额外的优化。有几种与SparkSQL交互的方法,包括SQL和DatasetAPI。计算结果时,将使用相同的执行引擎,这与用于表示计算的API/语言无关。这种统一意味着开发人员可以轻松地在不同的API之间来回切换,基于API的切换提供了表示给定转换的最自然的方式

  • 结构化数据是什么?
    存储在关系型数据库中的数据,就是结构化数据。
  • 半结构化数据是什么?
    类似xml、json等的格式的数据被称之为半结构化数据。
  • 非结构化数据是什么?
    音频、视频、图片等为非结构化数据。
    换句话说,SparkSQL处理的就是【二维表数据】。

1.SparkSQL特点

1.1易整合

在这里插入图片描述

1.2统一的数据访问

使用相同的连接方式连接不同的数据源
在这里插入图片描述

1.3兼容Hive

在已有的仓库上直接运行SQL或HQL
在这里插入图片描述

1.4标准的数据连接

采用JDBC或者ODBC直接连接
在这里插入图片描述

2 SparkSQL编程模型=DataFrame=DataSet

  • 通过两种方式操作SparkSQL,一种就是SQL,一种就是DataFrame和DataSet。

2.1 SQL

SQL操作的是表,所以要想用SQL进行操作,就需要把SparkSQL对应的编程模型转化为一张表才可以。

2.2 DataFrame是什么

在Spark中, DataFrame是一种以RDD为基础的分布式数据集,类似于传统数据库中的二维表格。DataFrame与RDD的主区别在于,前者带有schema元信息,即DataFrame所表示的二维表数据集的每一列都带有名称和类型。这使得Spark SOL得以洞察更多的结构信息,从而对藏于DataFrame背后的数据源以及作用于DataFrame之上的变换进行了针对性的优化,最终达到大幅提升运行时效率的目标。反观RDD, 由于无从得知所存数据元素的具体内部结构,SparkCore只能在stage层面进行简单、通用的流水线优化。
同时,与Hive类似,DataFrame也支持嵌套数据类型struct、array和map)。从API易用性的角度上看,DataFraneAPI提供的是一套高层的关系操作,比函数式的RDDAPI要更加友好,门槛更低。
RDD也是一张的二维表,不过没有表头,表名,字段,字段类型等信息。
DataFrame和DataSet是含有表头,表名,字段,字段类型的一张mysql中的二维表。
在这里插入图片描述
左侧的RDD[Person]虽然以Person为类型参数,但Spak框架本身不了解Person类的内部结构。而右侧的DataFrame却提供了详细的结构信息,使得Spark SQL 可以清楚地知道该数据集中包含哪些列,每列的名称和类型各是什么。

2.3 DataSet是什么

DataSet是分布式数据集合。DataSet是Spark1.6中添加的一个新抽象,是DataFrame的一个扩展。它提供了RDD的优势(强类型,使用强大的lambda函数的能力)以及SparkSQL优化执行引擎的优点。DataSet也可以使用功能性的转换(操作map, fatMap, filter等等)。

  • DataSet是DataFrameAPI的一个扩展,是SparkQL最新的数据抽象
  • 用户友好的API风格,既具有类型安全检查也具有DataFrame的查询优化特性;
  • 用样例类来对DataSet中定义数据的结构信息,样例类中每个属性的名称直接映射到DataSet中的字段名称;
  • DataSet是强类型的。比如可以有DataSet[Car], DataSet[Person]
  • DataFrame是DataSet的特列,DataFrame=DataSet[Row],所以可以通过as方法将DataFrame转换为DataSet。Row是一个类型,跟Car、Person这些的类型一样,所有的表结构信息都用Row来表示。获取数据时需要指定顺序。

一般的,将RDD称之为Spark体系中的第一代编程模型:DataFrame比RDD多了一个Schema元数据信息,被称之为Spark体系中的第二代编程模型:Dataset吸收了RDD的优点(强类型推断和强大的函数式编程)和DataFrame中的优化(SQL优化引擎,内存列存储),成为Spark的最新一代的编程模型

2.4 RDD,DataSet,DataFrame

  • RDD
    【弹性分布式数据集】,是Spark对数据进行的一种抽象,可以理解为Spark对数据的一种组织方式,更简单些说,RDD就是一种数据结构,里面包含了数据和操作数据的方法。
    从字面上就能看出的几个特点:
    • 弹性:
      • 数据可完全放内存或完全放磁盘,也可部分存放在内存,部分存放在磁盘,并可以自动切换
      • RDD出错后可自动重新计算(通过血缘自动容错)
      • 可checkpoint(设置检查点,用于容错),可persist或cache(缓存)里面的数据是分片的(也叫分区,partition),分片的大小可自由设置和细粒度调整
    • 分布式:
      • RDD中的数据可存放在多个节点上
    • 数据集:
      • 数据的集合

相对于与DataFrame和Dataset,RDD是Spark最底层的抽象,目前是开发者用的最多的,但逐步会转向DataFrame和Dataset(当然,这是Spark的发展趋势)

  • DataFrame
    DataFrame:理解了RDD,DataFrame就容易理解些,DataFrame的思想来源于Python的pandas库,RDD是一个数据集,DataFrame在RDD的基础上加了Schema 描述数据的信息,可以认为是元数据,DataFrame曾经就有个名字叫SchemaRDD)
    设RDD中的两行数据长这样:
1张三20
2李四21
3王五22

那么在DataFrame中数据变成这样:

ID:IntName:StringAge:Int
1张三20
2李四21
3王五22

从上面两个表格可以看出,DataFrame比RDD多了一个表头信息 (Schema),像一张表了,DataFrame还配套了新的操作数据的方法等,有了DataFrame这个高一层的抽象后,我们处理数据更加简单了,甚至可以用SQL来处理数据了,对开发者来说,易用性有了很大的提升,不仅如此,通过DataFrameAPI或SQL处理数据,会自动经过Spark优化器(Catalyst)的优化,即使你写的程序或SQL不高效,也可以运行的很快

  • DataSet
    相对于RDD,Dataset提供了强类型支持,也是在RDD的每行数据加了类型约束
    设RDD中的两行数据长这样:
1张三20
2李四21
3王五22

那么在DataFrame中数据变成这样:

ID:IntName:StringAge:Int
1张三20
2李四21
3王五22

那么在DataSet中数据变成这样:

Person(id:Int,Name:String,Age:Int)
Person(1,张三,20)
Person(2,李四,21)
Person(3,王五,22)

目前仅支持Scala、JavaAPI,尚未提供Python的API(所以一定要学习
Scala),相比DataFrame,Dataset提供了编译时类型检查,对于分布式程
序来讲,提交一次作业太费劲了(要编译、打包、上传、运行),到提交到集群运行时才发现错误,实在麻烦,这也是引入Dataset的一个重要原因。
使用DataFrame的代码json文件中并没有score字段,但是能编译通过,但是运行时会报异常,如下图代码所示:

 val df1 = spark.read.json("/tmp/people.json")
 //json文件中没有score字段,但是能编译通过
 val df2 = df1.filter("score>60").show()

而使用Dataset实现,会在IDE中报错,出错提前到了编译之前:

val ds1 = spark.read.json(("/tmp/people.json")).as[ People]
// 使用dataset这样写,在IDE中就能发现错误
val ds2 = ds1.filter(_.score < 60)
val ds3 = ds1.filter(_.age < 60)
// 打印
ds3.show()

3 SparkSQL核心编程

3.1 编程入口

SparkCore中,如果想要执行应用程序,需要首先构建上下文环境对象SpakContext,SparkSQL其实可以理解为对SparkCore的一种封装,不仅仅在模型上进行了封装,上下文环境对象也进行了封装。
在老的版本中,SparkSQL提供两种SQL查询起始点:一个叫SQLContext,用于Spark自己提供的SQL查询;一个HiveContext 用于连接Hive的查询。
SparkSession是Spark最新的SQL查询起始点,实质上是SQLContext和HiveContext的组合,所以在SQLContext和HiveContext上可用的API在SparkSession上同样是可以使用的。SparkSession内部封装了sparkContext ,所以计算实际上是由SpakContext完成的。
SparkSession的构建需要依赖SparkConf或者SparkContext。使用工厂构建器(Builder方式)模式创建SparkSession。

3.2 SparkSQL基本编程

3.2.1编程入口SparkSession
val session = SparkSession.builder
      .appName("test")  // 执行项目名称
      .master("local[*]") //选中本地执行方式
      // .enableHiveSupport() //开启支持Hive相关操作
      .getOrCreate() //创建session对象
3.2.2 DSL语法 -->结合SQL中关键字作为函数(算子)的名字传递参数进行编程方式–>接近于RDD编程
//无法读取表结构,优化为下行
//  val frame: DataFrame = session.read.json("data/people.json") 
    val frame: DataFrame = session.read.json
    (session.sparkContext.wholeTextFiles("data/people.json").values)
    //DSL语法 -->结合SQL中关键字作为函数(算子)的名字传递参数进行编程方式
    //        -->接近于RDD编程
    frame.printSchema()  //查看二维表结构
    /*
    运行结果:
    root
     |-- age: long (nullable = true)
     |-- height: double (nullable = true)
     |-- name: string (nullable = true)
     |-- province: string (nullable = true)
     */
    frame.show()  //相当于查看表中信息-->select * from people
    /*
    运行结果:
    +---+------+-------+--------+
    |age|height|   name|province|
    +---+------+-------+--------+
    | 10| 168.8|Michael|    广东|
    | 30| 168.8|   Andy|    福建|
    | 19| 169.8| Justin|    浙江|
    | 32| 188.8| 王启峰|    广东|
    | 10| 168.8|   John|    河南|
    | 19| 179.8|   Domu|    浙江|
    +---+------+-------+--------+
  */
    frame.show(3) //相当于查看表中前3行信息
    /*
    运行结果:
    +---+------+-------+--------+
    |age|height|   name|province|
    +---+------+-------+--------+
    | 10| 168.8|Michael|    广东|
    | 30| 168.8|   Andy|    福建|
    | 19| 169.8| Justin|    浙江|
    +---+------+-------+--------+
    only showing top 3 rows
   */
    //针对性查询某列数据-->select name,age from people
    frame.select("name","age").show()
    /*
    运行结果:
    +-------+---+
    |   name|age|
    +-------+---+
    |Michael| 10|
    |   Andy| 30|
    | Justin| 19|
    | 王启峰| 32|
    |   John| 10|
    |   Domu| 19|
    +-------+---+
   */
3.2.3 导入SparkSession中隐式转换操作,增强sql功能
import session.implicits._
    frame.select($"name",$"age").show()
    /*
    运行结果:
    +-------+---+
    |   name|age|
    +-------+---+
    |Michael| 10|
    |   Andy| 30|
    | Justin| 19|
    | 王启峰| 32|
    |   John| 10|
    |   Domu| 19|
    +-------+---+
   */
    //涉及到列运算时,每列都必须使用$符号
    //涉及到列运算时,每列也可以使用单引号字段名形式
    //等价于 select name,height-1,age+10 from people
    frame.select($"name",$"height"-1,'age+10).show()
    /*
   运行结果:
    +-------+------------+----------+
    |   name|(height - 1)|(age + 10)|
    +-------+------------+----------+
    |Michael|       167.8|        20|
    |   Andy|       167.8|        40|
    | Justin|       168.8|        29|
    | 王启峰|       187.8|        42|
    |   John|       167.8|        20|
    |   Domu|       178.8|        29|
    +-------+------------+----------+
   */
    //涉及到列运算时,也可以使用new Column方式
    //可以使用as修改列的别名
    //等价于select age+10 as age from people
    frame.select(new Column(name="age").+(10)).show()
    /*
    运行结果:
    +----------+
    |(age + 10)|
    +----------+
    |        20|
    |        40|
    |        29|
    |        42|
    |        20|
    |        29|
    +----------+
    */

    frame.select(new Column(name="age").+(10).as("age")).show()
    /*
    运行结果:
    +---+
    |age|
    +---+
    | 20|
    | 40|
    | 29|
    | 42|
    | 20|
    | 29|
    +---+
     */

    //分组聚合-->统计不用年龄的人数
    frame.select("age").groupBy("age").count().show()
    /*
   运行结果:
    +---+-----+
    |age|count|
    +---+-----+
    | 19|    2|
    | 32|    1|
    | 10|    2|
    | 30|    1|
    +---+-----+
    */

    //条件查询-->获取年龄超过18岁的
    frame.select("name","age","height").where("age>20").limit(4).show()
    /*
    运行结果:
    +------+---+------+
    |  name|age|height|
    +------+---+------+
    |  Andy| 30| 168.8|
    |王启峰| 32| 188.8|
    +------+---+------+
     */
3.2.4 SQL语法 -->直接写SQL或者HQL语言进行编程【算是SparkSQL主流】

注意:如果使用SQL的必要前提就是需要将数据转换为表

/*
    PS:创建表的参数为表名
    SQL语法操作中提供两种表:
    createOrReplaceTempView -->创建普通的临时表,作用域为当前session应用范围内有效
    createOrReplaceGlobalTempView -->创建普通的全局临时表,是当前application中可以使用,会覆盖原来数据
    createGlobalTempView --> 创建全局临时表,作用域为在整个当前application范围内有效,不会覆盖原来数据
    使用全局临时表时需要全路径访问:如global_temp.表名
    没有Replace关键字的global,不会覆盖,如创建,再创建,会报错
    有Replace关键字的global,会覆盖,如已经创建,再创建,会覆盖
     */
    frame.createGlobalTempView("people")
    session.sql(
      """
        |select * from global_temp.people
        |""".stripMargin).show()
    /*
    运行结果:
    +---+------+-------+--------+
    |age|height|   name|province|
    +---+------+-------+--------+
    | 10| 168.8|Michael|    广东|
    | 30| 168.8|   Andy|    福建|
    | 19| 169.8| Justin|    浙江|
    | 32| 188.8| 王启峰|    广东|
    | 10| 168.8|   John|    河南|
    | 19| 179.8|   Domu|    浙江|
    +---+------+-------+--------+
     */
    frame.createOrReplaceGlobalTempView("people_2")
    frame.createOrReplaceTempView("people_1")  //这个操作比较常用
    session.sql(
      """
        |select
        |age,
        |count(1) as countz
        |from people_1
        |group by age
        |""".stripMargin).show()
    /*
    运行结果:
    +---+------+
    |age|countz|
    +---+------+
    | 19|     2|
    | 32|     1|
    | 10|     2|
    | 30|     1|
    +---+------+
     */

参考自:https://www.bilibili.com/video/BV1WA411273z?p=5&spm_id_from=pageDriver&vd_source=6bd7b38d1d3cdff6e483a47870f6d418

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

《VitePress 简易速速上手小册》第2章:Markdown 与页面创建(2024 最新版)

文章目录 2.1 Markdown 基础及扩展2.1.1 基础知识点解析2.1.2 重点案例&#xff1a;技术博客2.1.3 拓展案例 1&#xff1a;食谱分享2.1.4 拓展案例 2&#xff1a;个人旅行日记 2.2 页面结构与布局设计2.2.1 基础知识点解析2.2.2 重点案例&#xff1a;公司官网2.2.3 拓展案例 1&…

来看看投资界最关心的 Sora 几大问题

作者&#xff1a;苍何&#xff0c;前大厂高级 Java 工程师&#xff0c;阿里云专家博主&#xff0c;CSDN 2023 年 实力新星&#xff0c;土木转码&#xff0c;现任部门技术 leader&#xff0c;专注于互联网技术分享&#xff0c;职场经验分享。 &#x1f525;热门文章推荐&#xf…

嵌入式开发-STM32CUBEMX使用—基于STM32G431RBTx

嵌入式–基于STM32G431RBTX 1.利用STM32CUBEMX生成工程框架 2.利用STM32CUBEMX生成初始化代码文件 创建文件 选择外晶振 Clock Configuration配置 按如下数据配置 Project Manager配置 Code Generator 在进行如上配置后即可生成 运行 在运行前需要把启动文件加入Applicati…

游戏被攻击如何处理

游戏服务器被攻击对于开发者来说是很常见的事情&#xff0c;一般游戏刚上线或者日活比较高的时候&#xff0c;很容易成为攻击者的目标&#xff0c;为了利益&#xff0c;攻击者会利用大量肉鸡发起DDoS跟CC攻击&#xff0c;导致游戏服务器瘫痪、网络卡顿、或用户掉线等情况&#…

多个.C 文件关于全局变量如何使用

&#x1d649;&#x1d65e;&#x1d658;&#x1d65a;!!&#x1f44f;&#x1f3fb;‧✧̣̥̇‧✦&#x1f44f;&#x1f3fb;‧✧̣̥̇‧✦ &#x1f44f;&#x1f3fb;‧✧̣̥̇:Solitary_walk ⸝⋆ ━━━┓ - 个性标签 - &#xff1a;来于“云”的“羽球人”。…

uniapp运动课程健身打卡系统微信小程序

考虑到实际生活中在我来运动管理方面的需要以及对该系统认真的分析,将系统分为小程序端模块和后台管理员模块&#xff0c;权限按管理员和用户这两类涉及用户划分。 (a) 管理员&#xff1b;管理员使用本系统涉到的功能主要有&#xff1a;首页、个人中心、用户管理、课程类别管理…

【知识分享】自动化测试首选接口自动化?

在分层测试的“金字塔”模型中&#xff0c;接口测试属于第二层服务集成测试范畴。 相比UI自动化测试而言&#xff0c;接口自动化测试收益更大&#xff0c;且容易实现&#xff0c;维护成本低&#xff0c;有着更高的投入产出比。因此&#xff0c;项目开展自动化测试的首选一般为接…

计网day5

六 传输层 6.1 传输层概述 6.2 UDP协议 6.3 TCP协议 TCP连接管理&#xff1a; TCP可靠传输&#xff1a; TCP拥塞控制&#xff1a;

kafka命令行操作

kafka命令行操作&#xff1a; 配置环境变量&#xff1a; export JAVA_HOME/usr/lib/jvm/java-1.8.0-openjdk export JRE_HOME${JAVA_HOME}/jre export CLASSPATH.:${JAVA_HOME}/lib:${JRE_HOME}/lib:$CLASSPATH export JAVA_PATH${JAVA_HOME}/bin:${JRE_HOME}/bin export PATH…

总结Rabbitmq的六种模式

RabbitMQ六种工作模式 RabbitMQ是由erlang语言开发&#xff0c;基于AMQP&#xff08;Advanced Message Queue 高级消息队列协议&#xff09;协议实现的消息队列&#xff0c;它是一种应用程序之间的通信方法&#xff0c;消息队列在分布式系统开发中应用非常广泛。 RabbitMQ有六…

Spring最新核心高频面试题(持续更新)

1 什么是Spring框架 Spring框架是一个开源的Java应用程序开发框架&#xff0c;它提供了很多工具和功能&#xff0c;可以帮助开发者更快地构建企业级应用程序。通过使用Spring框架&#xff0c;开发者可以更加轻松地开发Java应用程序&#xff0c;并且可以更加灵活地组织和管理应…

强化学习(SQL)

SQL——soft Q-learning 这是一种在最大化期望累计奖励的基础上&#xff0c;最大化熵项的算法&#xff0c;即该算法的优化目标是累计奖励和熵&#xff08;entropy&#xff09;的和。 SQL的好处&#xff1a; ①探索性强&#xff1a;可以在多模态任务中找到最佳的方案&#xf…

【问题解决】删除node节点后如何把node节点重新加入

环境明细 docker版本&#xff1a; 25.0.3kubeadm 版本&#xff1a;v1.25.0 1 在master节点删除node节点 [rootk8s-master ~]# kubectl delete nodes k8s-node-02 node "k8s-node-02" deleted [rootk8s-master ~]# kubectl get nodes -o wide NAME STAT…

嵌入式学习-qt-Day2

嵌入式学习-qt-Day2 一、思维导图 二、作业 1.使用手动连接&#xff0c;将登录框中的取消按钮使用qt4版本的连接到自定义的槽函数中&#xff0c;在自定义的槽函数中调用关闭函数 2.将登录按钮使用qt5版本的连接到自定义的槽函数中&#xff0c;在槽函数中判断ui界面上输入的账…

大数据构建知识图谱:从技术到实战的完整指南

文章目录 大数据构建知识图谱&#xff1a;从技术到实战的完整指南一、概述二、知识图谱的基础理论定义与分类核心组成历史与发展 三、知识获取与预处理数据源选择数据清洗实体识别 四、知识表示方法知识表示模型RDFOWL属性图模型 本体构建关系提取与表示 五、知识图谱构建技术图…

简化与效率:亚信安慧AntDB在数据处理中的创新

AntDB统一超融架构将外部接口的统一化极大简化了数据采集与处理的流程。传统的数据采集必须分成不同的阶段&#xff0c;先将数据收集起来&#xff0c;再进行处理&#xff0c;过程繁琐且易出错。而AntDB统一超融架构则将这两个环节合二为一&#xff0c;使得数据的采集与处理成为…

Project_Euler-15 题解

Project_Euler-15 题解 题目 思路 一眼背包问题&#xff0c;再看一眼广度优先搜索&#xff0c;再看一眼排列组合&#xff0c;这里讲一讲排列组合的解法吧&#xff1a; 以 2 ∗ 2 2*2 2∗2的组合为例可以发现&#xff1a; 无论哪一种方法&#xff0c;从左上角到右下角总共需要…

Docker部署Java项目

打包 添加 <!-- 打包插件--> <build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins> </build> 通过执行以下命令进…

Python学习笔记——自定义函数(基础知识)

自定义函数非常简洁有效地实现了代码的复用&#xff0c;让程序编写、阅读、测试和修改变得更加容易。 下面记录Python自定义函数的使用。 1、定义函数&#xff1a; def describe_pet(pet_name,animal_typedog):显示宠物的信息print(f"\nI have a {animal_type}.")…

年度巨作!《2024全球量子计算产业发展展望》发布

2月20日&#xff0c;《2024量子计算产业发展展望》的中文版报告通过光子盒官方平台发布&#xff0c;英文版报告通过ICV官方平台发布。 英文版报告获取地址&#xff1a; https://www.icvtank.com/newsinfo/897610.html 在过去的一年里&#xff0c;光子盒与您一同见证了全球量子…