大数据——SparkSQL学习笔记

Spark

一、SparkSQL简介

Spark用来处理结构化数据的一个模块，它提供了两个编程抽象分别叫做DataFrame和DataSet，它们用于作为分布式SQL查询引擎（类似于Hive，为便于进行MapReduce操作而使用类SQL语句进行Spark操作）。

➢ 数据兼容方面 SparkSQL 不但兼容 Hive，还可以从 RDD、parquet 文件、JSON 文件中获取数据，未来版本甚至支持获取 RDBMS 数据以及 cassandra 等 NOSQL 数据；

➢ 性能优化方面 除了采取 In-Memory Columnar Storage、byte-code generation 等优化技术外、将会引进 Cost Model 对查询进行动态评估、获取最佳物理计划等等；

➢ 组件扩展方面 无论是 SQL 的语法解析器、分析器还是优化器都可以重新定义，进行扩展。

——DataFrame
在这里插入图片描述

在 Spark 中，DataFrame 是一种以 RDD 为基础的分布式数据集，类似于传统数据库中的二维表格。DataFrame 与 RDD 的主要区别在于，前者带有 schema 元信息，即 DataFrame 所表示的二维表数据集的每一列都带有名称和类型。这使得 Spark SQL 得以洞察更多的结构信息，从而对藏于 DataFrame 背后的数据源以及作用于 DataFrame 之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。反观 RDD，由于无从得知所存数据元素的具体内部结构，Spark Core 只能在 stage 层面进行简单、通用的流水线优化。——DataSet

Spark提供的新抽象，它提供了 RDD 的优势（强类型，使用强大的 lambda 函数的能力）以及 Spark SQL 优化执行引擎的优点。DataSet 也可以使用功能性的转换（操作 map，flatMap，filter 等等）。

二、DataFrame

Spark SQL 的 DataFrame API 允许我们使用 DataFrame 而不用必须去注册临时表或者生成 SQL 表达式。DataFrame API 既有 transformation 操作也有 action 操作。

基础SQL语法	作用
var df = spark.read().<可选：json(“path”)…> 或spark.read().format(“”).option(“”).load()	读取数据(默认读取数据源为Parquet格式)
df.createOrReplaceTempView(“表名”)	根据数据建立临时视图
spark.sql(“SQL语句”).show	查询并展示
df.write.format(“<可选：json…>”) .save(“路径”)	保存数据

——JSON文件

Spark SQL 能够自动推测 JSON 数据集的结构，并将它加载为一个 Dataset[Row]. 可以通过 SparkSession.read.json()去加载 JSON 文件。注意：Spark 读取的 JSON 文件不是传统的 JSON 文件，每一行都应该是一个 JSON 串。

——CSV文件

Spark SQL 可以配置 CSV 文件的列表信息，读取 CSV 文件,CSV 文件的第一行设置为数据列。

——MySql文件

Spark SQL 可以通过 JDBC 从关系型数据库中读取数据的方式创建 DataFrame，通过对 DataFrame 一系列的计算后，还可以将数据再写回关系型数据库中。

val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
//创建 SparkSession 对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
import spark.implicits._
//方式 1：通用的 load 方法读取
spark.read.format("jdbc").option("url", "jdbc:mysql://linux1:3306/spark-sql").option("driver", "com.mysql.jdbc.Driver").option("user", "root").option("password", "123123").option("dbtable", "user").load().show
//释放资源
spark.stop()
/** 导出数据 */
case class User2(name: String, age: Long)
val rdd: RDD[User2] = spark.sparkContext.makeRDD(List(User2("lisi", 20), User2("zs", 30)))
val ds: Dataset[User2] = rdd.toDS
//方式 1：通用的方式 format 指定写出类型
ds.write.format("jdbc").option("url", "jdbc:mysql://linux1:3306/spark-sql").option("user", "root").option("password", "123123").option("dbtable", "user").mode(SaveMode.Append).save()
//释放资源
spark.stop()

DataFrame提供一种特定领域语言（domain-specific language，DSL）去管理结构化数据。从而不必再建立临时视图。

基础DSL语法	作用
df.rdd() rdd.toDF()	RDD与DataFrame互相转化
ds.rdd() rdd.toDS()	RDD与DataSet互相转化
ds.toDF() df.as(“类名”)	DataSet与DataFrame互相转化
df.printSchema	查看DataFrame信息
df.select(“属性名”).show	查询并展示
df.filter.($“属性名” = 条件).show	条件过滤
df.groupby(“属性名”).count.show	分组并计次

三、DataSet

DataSet是具有强类型的数据集合(DataFrame只有基础数据类型，没有自定义数据类型)，需要提供相应的类型信息。

基础语法	作用
case class Person(name: String, age: Long)	建立类
val caseClassDS = Seq(Person(“zhangsan”,2)).toDS()	使用样例类序列建立DataSet
caseClassDS.show	查看DataSet

——用户自定义函数(UDF)

sparksession.udf.register("prefixName", (name:String)=> {	//实现name前加字符"Name:" + name 
})
sparksession.sql("select age, prefixName(username) from user").show

——用户自定义聚合函数(UDAF)

强类型的 Dataset 和弱类型的 DataFrame 都提供了相关的聚合函数，如 count()， countDistinct()，avg()，max()，min()。除此之外，用户可以设定自己的自定义聚合函数。通过继承 UserDefinedAggregateFunction 来实现用户自定义弱类型聚合函数。从 Spark3.0 版本后，UserDefinedAggregateFunction 已经不推荐使用了。可以统一采用强类型聚合函数 Aggregator。

// TODO 创建 UDAF 函数
val udaf = new MyAvgAgeUDAF
// TODO 注册到 SparkSQL 中
spark.udf.register("avgAge", functions.udaf(udaf))
// TODO 在 SQL 中使用聚合函数
// 定义用户的自定义聚合函数
spark.sql("select avgAge(age) from user").show
// **************************************************
case class Buff( var sum:Long, var cnt:Long )
// totalage, count
class MyAvgAgeUDAF extends Aggregator[Long, Buff, Double]{override def zero: Buff = Buff(0,0)//聚合override def reduce(b: Buff, a: Long): Buff = {b.sum += ab.cnt += 1b}//合并多个缓冲区override def merge(b1: Buff, b2: Buff): Buff = {b1.sum += b2.sumb1.cnt += b2.cntb1}//结果输出override def finish(reduction: Buff): Double = {reduction.sum.toDouble/reduction.cnt}override def bufferEncoder: Encoder[Buff] = Encoders.productoverride def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

四、SparkSQL连接Hive

打开外部已经配置完成的Hive

➢ Spark 要接管 Hive 需要把 hive-site.xml 拷贝到 conf/目录下

➢ 把 Mysql 的驱动 copy 到 jars/目录下

➢ 如果访问不到 hdfs，则需要把 core-site.xml 和 hdfs-site.xml 拷贝到 conf/目录下

➢ 重启 spark-shell

使用IDEA实现SparkSQL连接Hive

➢ 导入jar包

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_2.12</artifactId><version>3.0.0</version>
</dependency>
<dependency><groupId>org.apache.hive</groupId><artifactId>hive-exec</artifactId><version>1.2.1</version>
</dependency>
<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.27</version>
</dependency>