以下笔记基于对尚硅谷spark教程的学习,Spark版本3.0
DataFrame是什么
在 Spark 中，DataFrame 是一种以 RDD 为基础的分布式数据集，类似于传统数据库中的二维表格。
DataFrame 与 RDD 的主要区别在于，前者带有 schema 元信息，即 DataFrame所表示的二维表数据集的每一列都带有名称和类型。
这使得 Spark SQL 得以洞察更多的结构信息，从而对藏于 DataFrame 背后的数据源以及作用于 DataFrame 之上的变换进行了针对性的优化，最终达到大幅提升运行时效率的目标。
反观 RDD，由于无从得知所存数据元素的具体内部结构，Spark Core 只能在 stage 层面进行简单、通用的流水线优化。
同时，与 Hive 类似，DataFrame 也支持嵌套数据类型（struct、array 和 map）。
从 API易用性的角度上看，DataFrame API 提供的是一套高层的关系操作，比函数式的 RDD API 要更加友好，门槛更低。
DataFrame 也是懒执行的，但性能上比 RDD 要高，主要原因：优化的执行计划，即查询计划通过 Spark catalyst optimiser 进行优化。
DataSet是什么
DataSet 是分布式数据集合。DataSet 是 Spark 1.6 中添加的一个新抽象，是 DataFrame的一个扩展。
它提供了 RDD 的优势（强类型，使用强大的 lambda 函数的能力）以及 Spark SQL 优化执行引擎的优点。
DataSet 也可以使用功能性的转换（操作 map，flatMap，filter等等）。
➢ DataSet 是 DataFrame API 的一个扩展，是 SparkSQL 最新的数据抽象
➢ 用户友好的 API 风格，既具有类型安全检查也具有 DataFrame 的查询优化特性；
➢ 用样例类来对 DataSet 中定义数据的结构信息，样例类中每个属性的名称直接映射到DataSet 中的字段名称；
➢ DataSet 是强类型的。比如可以有 DataSet[Car]，DataSet[Person]。
➢ DataFrame 是 DataSet 的特列，DataFrame=DataSet[Row] ，所以可以通过 as 方法将DataFrame 转换为 DataSet。
Row 是一个类型，跟 Car、Person 这些的类型一样，所有的表结构信息都用 Row 来表示。获取数据时需要指定顺序
DataFrame
SQL 语法
// 读取 JSON 文件创建 DataFrame
val df = spark.read.json("data/user.json")
// 对 DataFrame 创建一个临时表
df.createOrReplaceTempView("people")
// 通过 SQL 语句实现查询全表
val sqlDF = spark.sql("SELECT * FROM people")
// 结果展示
sqlDF.show
// 对于 DataFrame 创建一个全局表,普通临时表是 Session 范围内的，如果想应用范围内有效，可以使用全局临时表。使用全局临时表时需要全路径访问，如：global_temp.people
df.createGlobalTempView("people")
// 通过 SQL 语句实现查询全表
spark.sql("SELECT * FROM global_temp.people").show()
spark.newSession().sql("SELECT * FROM global_temp.people").show()
DSL 语法
DataFrame 提供一个特定领域语言(domain-specific language, DSL)去管理结构化的数据。
可以在 Scala, Java, Python 和 R 中使用 DSL，使用 DSL 语法风格不必去创建临时视图了
// 创建一个 DataFrame
val df = spark.read.json("data/user.json")
// 查看 DataFrame 的 Schema 信息
df.printSchema
// 只查看"username"列数据
df.select("username").show()
// 查看"username"列数据以及"age+1"数据
// 涉及到运算的时候, 每列都必须使用$, 或者采用引号表达式：单引号+字段名
df.select($"username",$"age" + 1).show
df.select('username, 'age + 1).show()
df.select('username, 'age + 1 as "newage").show()
// 查看"age"大于"30"的数据
df.filter($"age">30).show
// 按照"age"分组，查看数据条数
df.groupBy("age").count.show
RDD 转换为 DataFrame
// 指定列名
val idRDD = sc.textFile("data/id.txt")
idRDD.toDF("id").show
// 如果指定了结构样例类，可以直接转
case class User(name:String, age:Int)
sc.makeRDD(List(("zhangsan",30), ("lisi",40))).map(t=>User(t._1, t._2)).toDF.show
DataFrame 转换为 RDD
val df = sc.makeRDD(List(("zhangsan",30), ("lisi",40))).map(t=>User(t._1, t._2)).toDF
val rdd = df.rdd
DataSet
创建 DataSet
// 使用样例类序列创建 DataSet
case class Person(name: String, age: Long)
val caseClassDS = Seq(Person("zhangsan",2)).toDS
caseClassDS.show
// 使用基本类型的序列创建 DataSet
val ds = Seq(1,2,3,4,5).toDS
ds.show
RDD 转换为 DataSet
// SparkSQL 能够自动将包含有 case 类的 RDD 转换成 DataSet，case 类定义了 table 的结构，case 类属性通过反射变成了表的列名。
// Case 类可以包含诸如 Seq 或者 Array 等复杂的结构。
case class User(name:String, age:Int)
sc.makeRDD(List(("zhangsan",30), ("lisi",49))).map(t=>User(t._1, t._2)).toDS
DataSet 转换为 RDD
// DataSet 其实也是对 RDD 的封装，所以可以直接获取内部的 RDD
case class User(name:String, age:Int)
val res11 = sc.makeRDD(List(("zhangsan",30), ("lisi",49))).map(t=>User(t._1, t._2)).toDS
val rdd = res11.rdd
rdd.collect
DataFrame 和 DataSet 转换
DataFrame 其实是 DataSet 的特例，所以它们之间是可以互相转换的。
// DataFrame 转换为 DataSet
case class User(name:String, age:Int)
val df = sc.makeRDD(List(("zhangsan",30), ("lisi",49))).toDF("name","age")
val ds = df.as[User]
// DataSet 转换为 DataFrame
val df2 = ds.toDF
RDD、DataFrame、DataSet 三者的关系
** 三者的共性**
➢ RDD、DataFrame、DataSet 全都是 spark 平台下的分布式弹性数据集，为处理超大型数据提供便利;
➢ 三者都有惰性机制，在进行创建、转换，如 map 方法时，不会立即执行，只有在遇到Action 如 foreach 时，三者才会开始遍历运算;
➢ 三者有许多共同的函数，如 filter，排序等;
➢ 在对 DataFrame 和 Dataset 进行操作许多操作都需要这个包:import spark.implicits._（在创建好 SparkSession 对象后尽量直接导入）
➢ 三者都会根据 Spark 的内存情况自动缓存运算，这样即使数据量很大，也不用担心会内存溢出
➢ 三者都有 partition 的概念
➢ DataFrame 和 DataSet 均可使用模式匹配获取各个字段的值和类型
** 三者的区别**
RDD
➢ RDD 一般和 spark mllib 同时使用
➢ RDD 不支持 sparksql 操作
DataFrame
➢ 与 RDD 和 Dataset 不同，DataFrame 每一行的类型固定为 Row，每一列的值没法直接访问，只有通过解析才能获取各个字段的值
➢ DataFrame 与 DataSet 一般不与 spark mllib 同时使用
➢ DataFrame 与 DataSet 均支持 SparkSQL 的操作，比如 select，groupby 之类，还能注册临时表/视窗，进行 sql 语句操作
➢ DataFrame 与 DataSet 支持一些特别方便的保存方式，比如保存成 csv，可以带上表头，这样每一列的字段名一目了然
DataSet
➢ Dataset 和 DataFrame 拥有完全相同的成员函数，区别只是每一行的数据类型不同。
DataFrame 其实就是 DataSet 的一个特例 type DataFrame = Dataset[Row]
➢ DataFrame 也可以叫 Dataset[Row],每一行的类型是 Row，不解析每一行究竟有哪些字段，
只能用上面提到的 getAS 方法或者共性中的第七条提到的模式匹配拿出特定字段。
而Dataset中，每一行是什么类型是不一定的，在自定义了 case class 之后可以很自由的获得每一行的信息
用户自定义函数
UDF
// 注册UDF
spark.udf.register("addName", (x:String)=> "Name:" + x)
// 使用UDF
spark.sql("select addName(name),age from people").show()
UDAF-弱类型
从 Spark3.0 版本后，弱类型的UserDefinedAggregateFunction已经不推荐使用了。
/*
自定义聚合函数类：计算年龄的平均值
1. 继承UserDefinedAggregateFunction
2. 重写方法(8)
*/
class MyAvgUDAF extends UserDefinedAggregateFunction{
// 输入数据的结构 : Int
override def inputSchema: StructType = {
StructType(
Array(
StructField("age", LongType)
)
)
}
// 缓冲区数据的结构 : Buffer
override def bufferSchema: StructType = {
StructType(
Array(
StructField("total", LongType),
StructField("count", LongType)
)
)
}
// 函数计算结果的数据类型：Out
override def dataType: DataType = LongType
// 函数的稳定性
override def deterministic: Boolean = true
// 缓冲区初始化
override def initialize(buffer: MutableAggregationBuffer): Unit = {
//buffer(0) = 0L
//buffer(1) = 0L
buffer.update(0, 0L)
buffer.update(1, 0L)
}
// 根据输入的值更新缓冲区数据
override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
buffer.update(0, buffer.getLong(0)+input.getLong(0))
buffer.update(1, buffer.getLong(1)+1)
}
// 缓冲区数据合并
override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
buffer1.update(0, buffer1.getLong(0) + buffer2.getLong(0))
buffer1.update(1, buffer1.getLong(1) + buffer2.getLong(1))
}
// 计算平均值
override def evaluate(buffer: Row): Any = {
buffer.getLong(0)/buffer.getLong(1)
}
}
// 在spark中注册聚合函数
spark.udf.register("ageAvg", new MyAvgUDAF())
// 使用自定义聚合函数
spark.sql("select avgAge(age) from user").show()
UDAF-强类型(SQL)
/*
自定义聚合函数类：计算年龄的平均值
1. 继承org.apache.spark.sql.expressions.Aggregator, 定义泛型
IN : 输入的数据类型 Long
BUF : 缓冲区的数据类型 Buff
OUT : 输出的数据类型 Long
2. 重写方法(6)
*/
case class Buff( var total:Long, var count:Long )
class MyAvgUDAF extends Aggregator[Long, Buff, Long]{
// z & zero : 初始值或零值
// 缓冲区的初始化
override def zero: Buff = {
Buff(0L,0L)
}
// 根据输入的数据更新缓冲区的数据
override def reduce(buff: Buff, in: Long): Buff = {
buff.total = buff.total + in
buff.count = buff.count + 1
buff
}
// 合并缓冲区
override def merge(buff1: Buff, buff2: Buff): Buff = {
buff1.total = buff1.total + buff2.total
buff1.count = buff1.count + buff2.count
buff1
}
//计算结果
override def finish(buff: Buff): Long = {
buff.total / buff.count
}
// 缓冲区的编码操作
override def bufferEncoder: Encoder[Buff] = Encoders.product
// 输出的编码操作
override def outputEncoder: Encoder[Long] = Encoders.scalaLong
}
// 在spark中注册聚合函数
spark.udf.register("ageAvg", functions.udaf(new MyAvgUDAF()))
// 使用自定义聚合函数
spark.sql("select ageAvg(age) from user").show
UDAF-强类型(DSL)
/*
自定义聚合函数类：计算年龄的平均值
1. 继承org.apache.spark.sql.expressions.Aggregator, 定义泛型
IN : 输入的数据类型 User
BUF : 缓冲区的数据类型 Buff
OUT : 输出的数据类型 Long
2. 重写方法(6)
*/
case class User(username:String, age:Long)
case class Buff( var total:Long, var count:Long )
class MyAvgUDAF extends Aggregator[User, Buff, Long]{
// z & zero : 初始值或零值
// 缓冲区的初始化
override def zero: Buff = {
Buff(0L,0L)
}
// 根据输入的数据更新缓冲区的数据
override def reduce(buff: Buff, in: User): Buff = {
buff.total = buff.total + in.age
buff.count = buff.count + 1
buff
}
// 合并缓冲区
override def merge(buff1: Buff, buff2: Buff): Buff = {
buff1.total = buff1.total + buff2.total
buff1.count = buff1.count + buff2.count
buff1
}
//计算结果
override def finish(buff: Buff): Long = {
buff.total / buff.count
}
// 缓冲区的编码操作
override def bufferEncoder: Encoder[Buff] = Encoders.product
// 输出的编码操作
override def outputEncoder: Encoder[Long] = Encoders.scalaLong
}
// 将UDAF函数转换为查询的列对象
val udafCol: TypedColumn[User, Long] = new MyAvgUDAF().toColumn
// 使用自定义聚合函数
ds.select(udafCol).show
数据的加载和保存
通用的加载和保存方式
SparkSQL 默认读取和保存的文件格式为 parquet，修改配置项 spark.sql.sources.default，可修改默认数据源格式
// 加载数据
spark.read.format("…")[.option("…")].load("…")
// format("…")：指定加载的数据类型，包括"csv"、"jdbc"、"json"、"orc"、"parquet"和"textFile"
// load("…")：在"csv"、"jdbc"、"json"、"orc"、"parquet"和"textFile"格式下需要传入加载数据的路径
// option("…")：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable
// 保存数据
ds.write[.mode("…")].format("…")[.option("…")].save("…")
// format("…")：指定保存的数据类型，包括"csv"、"jdbc"、"json"、"orc"、"parquet"和"textFile"
// save ("…")：在"csv"、"orc"、"parquet"和"textFile"格式下需要传入保存数据的路径
// option("…")：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable
// 保存操作可以使用 SaveMode, 用来指明如何处理数据，使用 mode()方法来设置
Scala/JavaAny LanguageMeaningSaveMode.ErrorIfExists(default)"error"(default)如果文件已经存在则抛出异常SaveMode.Append"append"如果文件已经存在则追加SaveMode.Overwrite"overwrite"如果文件已经存在则覆盖SaveMode.Ignore"ignore"如果文件已经存在则忽略
特定格式的文件读写
// 读取csv文件,返回格式DataFrame
spark.read[.option("…")].csv("…")
spark.read.format("csv").option("sep", ";").option("inferSchema", "true").option("header", "true").load("data/user.csv")
// 读取json文件,返回格式DataFrame
spark.read[.option("…")].json("…")
// 读取orc文件,返回格式DataFrame
spark.read[.option("…")].orc("…")
// 读取parquet文件,返回格式DataFrame
spark.read[.option("…")].parquet("…")
// 读取text文件,返回格式DataFrame,第一个字段名为value包含文件行内容,如果有分区,第二个字段为分区列
spark.read[.option("…")].text("…")
// 读取text文件,返回格式Dataset[String],内部实际调用text()后再转换
spark.read[.option("…")].textFile("…")
// 保存csv文件
ds.write[.mode("…")][.option("…")].csv("…")
// 保存json文件
ds.write[.mode("…")][.option("…")].json("…")
// 保存orc文件
ds.write[.mode("…")][.option("…")].orc("…")
// 保存parquet文件
ds.write[.mode("…")][.option("…")].parquet("…")
// 保存text文件
ds.write[.mode("…")][.option("…")].text("…")
特殊方式读取
我们也可以直接在文件上进行查询: 文件格式.`文件路径`
spark.sql("select * from json.`/opt/module/data/user.json`").show
JDBC读写
// 读取方式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
// 读取方式2:通用的 load 方法读取 参数另一种形式
spark.read.format("jdbc")
.options(Map("url"->"jdbc:mysql://linux1:3306/spark-sql?user=root&password=123123",
"dbtable"->"user",
"driver"->"com.mysql.jdbc.Driver"))
.load().show
// 读取方式3:使用 jdbc 方法读取
val props: Properties = new Properties()
props.setProperty("user", "root")
props.setProperty("password", "123123")
val df: DataFrame = spark.read.jdbc("jdbc:mysql://linux1:3306/spark-sql", "user", props)
df.show
// 保存方式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()
// 保存方式2：通过 jdbc 方法
val props: Properties = new Properties()
props.setProperty("user", "root")
props.setProperty("password", "123123")
ds.write.mode(SaveMode.Append).jdbc("jdbc:mysql://linux1:3306/spark-sql", "user", props)
Hive
** spark-shell访问内嵌的HIVE**
如果使用 Spark 内嵌的 Hive, 则什么都不用做, 直接使用即可
Hive的元数据存储在derby中, 默认仓库地址:$SPARK_HOME/spark-warehouse
如果你的classpath中有配好的hdfs-site.xml，默认的文件系统就是HDFS，否则就是本地文件系统
** spark-shell访问外部的 HIVE**
Spark 要接管 Hive 需要把 hive-site.xml 拷贝到 conf/目录下
把Mysql的驱动copy到jars/目录下
如果访问不到hdfs，则需要把core-site.xml和hdfs-site.xml拷贝到conf/目录下
重启 spark-shell
** 代码操作Hive**
导入依赖
<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>
将 hive-site.xml 文件拷贝到项目的 resources 目录中
//创建 SparkSession
val spark: SparkSession = SparkSession
.builder()
.enableHiveSupport()
.master("local[*]")
.appName("sql")
.getOrCreate()
如果在执行操作时报错用户没有权限，则在代码前面加上
System.setProperty("HADOOP_USER_NAME", "hive");
版权归原作者 langzitianya 所有， 如有侵权，请联系我们删除。