SparkSQL
01.快速入门
什么是SparkSQL
SparkSQL是Spark的一个模块,用于处理海量结构化数据
为什么学习SparkSQL
SparkSQL是非常成熟的海量结构化数据处理框架:
学习SparkSQL主要在2个点:
- SparkSQL本身十分优秀,支持SQL语言、性能强、可以自动优化、API简单、兼容HIVE等等
- 企业大面积在使用SparkSQL处理业务数据
1、离线开发
2、数仓搭建
3、科学计算
4、数据分析
SparkSQL的特点
- 融合性:SQL可以无缝集成在代码中,随时用SQL处理数据
- 统一数据访问:一套标准API可读写不同数据源
- Hive兼容:可以使用SparkSQL直接计算并生成Hive数据表
- 标准化连接:支持标准化JDBC/ODBC连接,方便和各种数据源进行数据交互
SparkSQL发展历史-前身Shark框架
SparkSQL发展历史
02.SparkSQL概述
SparkSQL和Hive的异同
相同点:
1、分布式SQL计算引擎
2、构建大规模结构化数据计算的绝佳利器,同时SparkSQL拥有更好的性能
不同点:
SparkSQL的数据抽象
1、SparkSQL-DataFrame
- 二维表数据结构
- 分布式结构集合(分区)
2、SparkSQL FOor JVM-DataSet【可用于Java\Scala\语言】
3、SparkSQL For Python\R-DataFrame【可用于Java\Scale\Python\R】
DataFrame概述
DataFrame是按照二维表格的形式存储数据
RDD则是存储对象本身
SparkSession对象
在RDD阶段,程序的执行入口对象是SparkContext
在Sparke2.0后,推出SparkSeaaion对象,作为Spark编码的统一入口对象
SparkSession对象可以:
1、用于SparkSQL编程作为入口对象
2、用于SparkCore编程,可以通过SparkSession对象中获取到SparkContext
03.DataFrame入门和操作
DataFrame的组成
DataFrame是一个二维表结构,那么表格结构就有无法绕开的三个点:
-
行
-
列
-
表结构表述
比如MySQL中的一张表: -
由许多行组成
-
数据也可以被分成多个列
-
表也有表结构信息(列、列名、列类型、列约束等)
在结构层面上: -
StructType对象描述整个DataFrame的表结构
-
StructField对象描述一个列的信息
在数据层面上: -
Row对象记录一行数据
-
Column对象记录一列数据并包含列的信息
在表结构层面,DataFrame的表结构由:
StructType描述:
struct_type = StructType().\
add("id",IntegerType(),False).\
add("name",StringType(),True).\
add("age",IntegerType(),False)
一个StructField记录:列名、列类型、列是否运行为空
多个StructField组成一个StructedType对象
一个StructType对象可以描述一个DataFrame:有几个列、每个列的名字和类型、每个列是否为空
一行数据描述为Row对象,如Row(1,张三,11)
一列数据描述为Column对象,Column对象包含一列数据和列的信息
DataFrame的代码构建-基于RDD-1
#coding:utf8
from pyspark.sql import SparkSession
if __name__ == '__main__':
# 0.构建执行环境入口对象SparkSession
spark = SparkSesion.builder.\
appName("test").\
master("local[*]").\
getOrcreate()
#1.基于RDD转换成DataFrame
sc = spark.sparkContext
rdd = sc.testFile("../data/input/sql/people.txt).map(lambda x:x.spalit(",")).map(lambda x:(x[0],int(x[1])))
#2.构建DataFrame对象
## 参数一:被转换的rdd
## 参数二:指定列名,通过list的形式指定,按照顺序依次提供字符串名称即可
df = spark.createDataFrame(rdd,schema=['name','age'])
df.printSchema()
## 参数一:表示 展示出来多少条数据,默认不传的话是20
## 参数二:表示是否对列进行截断,如果列的数据长度超过20个字符串长度,后续的内容不显示以。。。代替
## 如果给False,表示不截断全部显示,默认是True
df.show(20,False)
DataFrame的代码构建-基于RDD-2
#coding:utf8
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType,StringType,IntegerType
if __name__ == '__main__':
# 0.构建执行环境入口对象SparkSession
spark = SparkSesion.builder.\
appName("test").\
master("local[*]").\
getOrcreate()
#1.基于RDD转换成DataFrame
sc = spark.sparkContext
rdd = sc.testFile("../data/input/sql/people.txt).map(lambda x:x.spalit(",")).map(lambda x:(x[0],int(x[1])))
schema=StructType().add('name',StringType(),True).add('age'IntegerType(),False)
df = spark.createDataFrame(rdd,schema)
df.printSchema()
## 参数一:表示 展示出来多少条数据,默认不传的话是20
## 参数二:表示是否对列进行截断,如果列的数据长度超过20个字符串长度,后续的内容不显示以。。。代替
## 如果给False,表示不截断全部显示,默认是True
df.show(20,False)
DataFrame的代码构建-基于RDD-3
该方法用于对数据类型不敏感
#coding:utf8
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType,StringType,IntegerType
if __name__ == '__main__':
# 0.构建执行环境入口对象SparkSession
spark = SparkSesion.builder.\
appName("test").\
master("local[*]").\
getOrcreate()
#1.基于RDD转换成DataFrame
sc = spark.sparkContext
rdd = sc.testFile("../data/input/sql/people.txt).map(lambda x:x.spalit(",")).map(lambda x:(x[0],int(x[1])))
# toDF的方式构建DataFrame
df1 = rdd.toDF(['name','age'])
# 方法二
schema=StructType().add('name',StringType(),True).add('age'IntegerType(),False)
rdd.toDF(schema)
df1.printSchema()
## 参数一:表示 展示出来多少条数据,默认不传的话是20
## 参数二:表示是否对列进行截断,如果列的数据长度超过20个字符串长度,后续的内容不显示以。。。代替
## 如果给False,表示不截断全部显示,默认是True
df1.show(20,False)
DataFrame的代码构建-基于Pandas的DataFrame
#coding:utf8
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType,StringType,IntegerType
if __name__ == '__main__':
# 0.构建执行环境入口对象SparkSession
spark = SparkSesion.builder.\
appName("test").\
master("local[*]").\
getOrcreate()
#1.基于RDD转换成DataFrame
sc = spark.sparkContext
# 基于Pandas的DataFrame构建SparkSQL的DataFrame对象
pdf = pd.DataFrame({'id':[1,2,3],'name':['张大仙','王小小','王大锤'],'age':[11,11,11]})
# 将Pandas的DF对象转换成SparkDF
df1 = spark.createDataFrame(pdf)
df1.printSchema()
## 参数一:表示 展示出来多少条数据,默认不传的话是20
## 参数二:表示是否对列进行截断,如果列的数据长度超过20个字符串长度,后续的内容不显示以。。。代替
## 如果给False,表示不截断全部显示,默认是True
df1.show(20,False)
DataFrame的代码构建-读取外部数据-text
构建StructType,text数据源,读取数据的特点是,是将一整行只作为一个列读取,默认列名是value 类型是String
spark session.read.format(“text|csv|json|parquet|orc|avro|jdbc…”)
.option(“k”,“v”)#option可选
.schema(StructType|String)#STRING的语法如。Schema(“name STRING”,“age INT” )
.load(“被读取文件的路径,支持本地文件系统和HDFS”)
#coding:utf8
from pyspark.sql import SparkSession
if __name__ == '__main__':
# 0.构建执行环境入口对象SparkSession
spark = SparkSesion.builder.\
appName("test").\
master("local[*]").\
getOrcreate()
#1.基于RDD转换成DataFrame
sc = spark.sparkContext
# 构建StructType,text数据源,读取数据的特点是,是将一整行只作为一个列读取,默认列名是value 类型是String
schema = StructType().add('data',StirngType(),True)
df = spark.read.format('text').schema(schema=schema).load('../data/input/sql/people.txt')
df.printSchema()
## 参数一:表示 展示出来多少条数据,默认不传的话是20
## 参数二:表示是否对列进行截断,如果列的数据长度超过20个字符串长度,后续的内容不显示以。。。代替
## 如果给False,表示不截断全部显示,默认是True
df.show(20,False)
DataFrame的代码构建-读取外部数据-json
json类型自带有Schema信息
#coding:utf8
from pyspark.sql import SparkSession
if __name__ == '__main__':
# 0.构建执行环境入口对象SparkSession
spark = SparkSesion.builder.\
appName("test").\
master("local[*]").\
getOrcreate()
#1.基于RDD转换成DataFrame
sc = spark.sparkContext
# json类型自带有Schema信息
schema = StructType().add('data',StirngType(),True)
df = spark.read.format('json').load('../data/input/sql/people.txt')
df.printSchema()
## 参数一:表示 展示出来多少条数据,默认不传的话是20
## 参数二:表示是否对列进行截断,如果列的数据长度超过20个字符串长度,后续的内容不显示以。。。代替
## 如果给False,表示不截断全部显示,默认是True
df.show(20,False)
DataFrame的代码构建-读取外部数据-csv
#coding:utf8
from pyspark.sql import SparkSession
if __name__ == '__main__':
# 0.构建执行环境入口对象SparkSession
spark = SparkSesion.builder.\
appName("test").\
master("local[*]").\
getOrcreate()
#1.基于RDD转换成DataFrame
sc = spark.sparkContext
# json类型自带有Schema信息
schema = StructType().add('data',StirngType(),True)
df = spark.read.format('csv').\
option('sep',';').\
option('header',True).\
option('encoding','utf-8').\
schema('name STRING age INT,job STRING').\
load('../data/input/sql/people.txt')
df.printSchema()
df.show(20,False)
DataFrame的代码构建-读取外部数据-parquet
parquet:是spark中常用的一种列式存储文件格式,和Hive中ORC差不多,他俩都是列存储格式
parquet对比普通文本文件的区别
- parquet内置schema(列名、列类型、是否为空)
- 存储是以列作为存储格式
- 存储时序列化存储在文件中的,有压缩属性体积小
#coding:utf8
from pyspark.sql import SparkSession
if __name__ == '__main__':
# 0.构建执行环境入口对象SparkSession
spark = SparkSesion.builder.\
appName("test").\
master("local[*]").\
getOrcreate()
#1.基于RDD转换成DataFrame
sc = spark.sparkContext
# parquet类型自带有Schema信息
schema = StructType().add('data',StirngType(),True)
df = spark.read.format('parquet').load('../data/input/sql/people.txt')
df.printSchema()
df.show(20,False)
DataFrame的入门操作
DataFrame支持两种风格进行编程,分别是:
- DSL风格:DataFrame的特有API,调用API的方式来处理Data
#coding:utf8
from pyspark.sql import SparkSession
if __name__ == '__main__':
# 0.构建执行环境入口对象SparkSession
spark = SparkSesion.builder.\
appName("test").\
master("local[*]").\
getOrcreate()
#1.基于RDD转换成DataFrame
sc = spark.sparkContext
# parquet类型自带有Schema信息
df = spark.read.format('csv').load('../data/input/sql/people.txt')
# column对象的获取
id_column = df['id']
subject_column = df['subject']
# DSL风格演示
df.select(["id","subject"]).show()
df.select ("id","subject").show()
df.select(id_column,subject_column)
# filter API
df.filter("score < 99").show()
df.filter(df['score'] < 99).show()
# where API
df.where("score < 99").show()
df.where(df['score'] < 99).show()
# group by API
df.groupBy("subject").count().show()
df.groupBy(df['subject']).count().show()
- SQL风格:spark.sql(“select * from XXX”)
使用sparj.sql()来执行SQL语句查询,结果返回一个DataFrame
df.createTempView("score") #注册一个临时视图
df.createOrReplaceTempView("socre") #注册一个临时表,如果存在,进行替换
df.createGlobalTempView("score") # 注册一个全局表
全局表:跨sparksession对象使用,在一个程序内的多个sparkSession中均可调用,查询前带上前缀
global_temp.
SparkSQL数据清洗API
- 去重方法:dropDuplicates
- 缺失值处理:
- dropna 是可以对缺失值进行删除;只要列中有null 就删除这一行数据
参数:thread=3表示,至少满足3个有效列,不满足就删除当前数据 - fillna(“loss”) 对缺失值的列进行填充
- fillna(“N/A”,subset=[‘job’])指定列进行填充
- fillna({‘name’:‘未知姓名’,‘age’:1,‘job’:‘worker’})设定一个字典,对所有的列提供填充规则
- dropna 是可以对缺失值进行删除;只要列中有null 就删除这一行数据
DataFrame数据写出
SparkSQL 统一API写出DataFrame数据
df.write.mode().format().option(K,V).save(PATH)
- mode,传入模式字符串可选:append追加,overwrite覆盖,ignore忽略,error重复就报异常(默认的)
- format,传入格式字符串,可选:text,csv,json,parquet,orc,avro,jdbc
- save 写出的路径,支持本地文件和HDFS
04.SparkSQL函数定义
SparkSQL定义UDF
pyspark UDF
SparkSQL使用窗口函数
-
聚合开窗函数
-
排序开窗函数
– ROW_NUMBER() OVER()
–DENSE_RANK() OVER()
–RANK() OVER() -
NTILE分组窗口
05.SparkSQL的运行流程
SparkSQL的自动优化
RDD的运行完全是按照开发者的代码执行,如果开发者水平有限,RDD的执行效率也会收到影响
而SparkSQL会对写完的代码,执行“自动优化”,以提升代码运行效率,避免开发者水平影响到代码执行效率;依赖于:Catalyst优化器
Catalyst优化器
STEP1:解析SQL,并生成AST(抽象语法树)
大方面的优化点有2个:
- 谓词下推、断言下推:将逻辑判断提前到前面,以减少shuffle阶段的数据量
- 列值剪裁:将加载的列进行剪裁,尽量减少被处理数据的宽度
SparkSQL的执行流程
06.SparkSQL整合Hive
Hive执行流程
SparkOn Hive
Spark On Hive就是因为Spark自身没有元数据管理功能,所以使用Hive的Metastore服务做为元数据管理服务。计算有Spark执行文章来源:https://www.toymoban.com/news/detail-808280.html
07.分布式SQL引擎配置
文章来源地址https://www.toymoban.com/news/detail-808280.html
到了这里,关于【spark】SparkSQL的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!