3. SparkSQL数据的加载与保存
3.1 通用的加载和保存方式
spark.read.load 是加载数据的通用方法
df.write.save 是保存数据的通用方法
3.1.1 加载数据
1）read直接加载数据
spark.read.json("/opt/module/spark-local/people.json")
2）format指定加载数据类型
spark.read.format("json").load("/opt/module/spark-local/people.json")
format("…")：指定加载的数据类型，包括"csv"、“jdbc”、“json”、“orc”、“parquet"和"textFile”
load("…")：在"csv"、“jdbc”、“json”、“orc”、“parquet"和"textFile"格式下需要传入加载数据的路径
option(”…")：在"jdbc"格式下需要传入JDBC相应参数，url、user、password和dbtable
3）在文件上直接运行SQL
spark.sql("select * from json.`/opt/module/spark-local/people.json`").show
输出：
3.1.2 保存数据
1）write直接保存数据
scala> df.write.
csv jdbc json orc parquet textFile… …
保存数据的相关参数需写到上述方法中。如：textFile需传入加载数据的路径，jdbc需传入JDBC相关参数。
# 默认保存格式为parquet
scala> df.write.save("/opt/module/spark-local/output")
# 可以指定为保存格式，直接保存，不需要再调用save了
scala> df.write.json("/opt/module/spark-local/output")
2）format指定保存数据类型
df.write.format("…")[.option("…")].save("…")
format("…")：指定保存的数据类型，包括"csv"、“jdbc”、“json”、“orc”、“parquet"和"textFile”。
save ("…")：在"csv"、“orc”、“parquet"和"textFile"格式下需要传入保存数据的路径。
option(”…")：在"jdbc"格式下需要传入JDBC相应参数，url、user、password和dbtable
3）文件保存选项
保存操作可以使用 SaveMode, 用来指明如何处理数据，使用mode()方法来设置。有一点很重要: 这些 SaveMode 都是没有加锁的, 也不是原子操作。SaveMode是一个枚举类，其中的常量包括：
| Scala/Java | Any | Language Meaning |
|---|---|---|
| SaveMode.ErrorIfExists(default) | “error”(default) | 如果文件已经存在则抛出异常 |
| SaveMode.Append | “append” | 如果文件已经存在则追加 |
| SaveMode.Overwrite | “overwrite” | 如果文件已经存在则覆盖 |
| SaveMode.Ignore | “ignore” | 如果文件已经存在则忽略 |
使用指定format指定保存类型进行保存
df.write.mode("append").json("/opt/module/spark-local/output")
3.1.3 默认数据源
Spark SQL的默认数据源为Parquet格式。数据源为Parquet文件时，Spark SQL可以方便的执行所有的操作，不需要使用format。修改配置项spark.sql.sources.default，可修改默认数据源格式。
1）加载数据
val df = spark.read.load("/opt/module/spark-local/examples/src/main/resources/users.parquet").show
2）保存数据
var df = spark.read.json("/opt/module/spark-local/people.json")
df.write.mode("append").save("/opt/module/spark-local/output")
3.2 JSON文件
Spark SQL 能够自动推测 JSON数据集的结构，并将它加载为一个Dataset[Row]. 可以通过SparkSession.read.json()去加载一个 一个JSON 文件。
1）导入隐式转换
import spark.implicits._
2）加载JSON文件
val path = "/opt/module/spark-local/people.json"
val peopleDF = spark.read.json(path)
3）创建临时表
peopleDF.createOrReplaceTempView("people")
4）数据查询
val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 AND 19")
teenagerNamesDF.show()
3.3 MySQL
Spark SQL可以通过JDBC从关系型数据库中读取数据的方式创建DataFrame，通过对DataFrame一系列的计算后，还可以将数据再写回关系型数据库中。
如果使用spark-shell操作，可在启动shell时指定相关的数据库驱动路径或者将相关的数据库驱动放到spark的类路径下。
bin/spark-shell
–jars mysql-connector-java-5.1.27-bin.jar
在Idea中通过JDBC对Mysql进行操作
3.3.1 通过jdbc对MySQL进行读写操作
/**
* 通过jdbc对MySQL进行读写操作
*/
object SparkSQL01_MySQL {
def main(args: Array[String]): Unit = {
//创建配置文件对象
val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL01_MySQL")
//创建SparkSession对象
val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
import spark.implicits._
//从MySql数据库中读取数据
//方式1 通用的load方法读取
spark.read.format("jdbc")
.option("url", "jdbc:mysql://hadoop102:3306/test")
.option("driver ", "com.mysql.jdbc.Driver")
.option("user", "root")
.option("password", "000000")
.option("dbtable", "user")
.load().show()
//方式2 通用的load方法读取 参数另一种形式
spark.read.format("jdbc")
.options(
Map(
"url" -> "jdbc:mysql://hadoop102:3306/test",
"driver" -> "com.mysql.jdbc.Driver",
"user" -> "root",
"password" -> "000000",
"dbtable" -> "user"
)
)
.load().show()
//方式3 使用jdbc方法读取
val properties: Properties = new Properties()
properties.setProperty("user", "root")
properties.setProperty("password", "000000")
spark.read.jdbc("jdbc:mysql://hadoop102:3306/test", "user", properties).show()
//向MySql数据库中写数据
val rdd: RDD[User] = spark.sparkContext.makeRDD(List(User("aaa", 20), User("bbb", 18)))
//将RDD转换为DF
val df: DataFrame = rdd.toDF()
//将RDD转换为DS
val ds: Dataset[User] = rdd.toDS()
ds.write.format("jdbc")
.option("url", "jdbc:mysql://hadoop102:3306/test")
.option("driver ", "com.mysql.jdbc.Driver")
.option("user", "root")
.option("password", "000000")
.option("dbtable", "user")
.mode(SaveMode.Append)
.save()
//释放资源
spark.stop()
}
}
case class User(name: String, age: Int)
输出：
读：
+---+--------+---+
| id| name|age|
+---+--------+---+
| 1| xuzhu| 20|
| 2| duanyu| 30|
| 3|qiaofeng| 40|
+---+--------+---+
写：
3.3.2 向JDBC写数据
3.4 Hive
Apache Hive 是 Hadoop 上的 SQL 引擎，Spark SQL编译时可以包含 Hive 支持，也可以不包含。
包含 Hive 支持的 Spark SQL 可以支持 Hive 表访问、UDF (用户自定义函数)以及 Hive 查询语言(HiveQL/HQL)等。需要强调的一点是，如果要在 Spark SQL 中包含Hive 的库，并不需要事先安装 Hive。一般来说，最好还是在编译Spark SQL时引入Hive支持，这样就可以使用这些特性了。如果你下载的是二进制版本的 Spark，它应该已经在编译时添加了 Hive 支持。
若要把 Spark SQL 连接到一个部署好的 Hive 上，你必须把 hive-site.xml 复制到 Spark的配置文件目录中($SPARK_HOME/conf)。即使没有部署好 Hive，Spark SQL 也可以运行，需要注意的是，如果你没有部署好Hive，Spark SQL 会在当前的工作目录中创建出自己的 Hive 元数据仓库，叫作 metastore_db。此外，对于使用部署好的Hive，如果你尝试使用 HiveQL 中的 CREATE TABLE (并非 CREATE EXTERNAL TABLE)语句来创建表，这些表会被放在你默认的文件系统中的 /user/hive/warehouse 目录中(如果你的 classpath 中有配好的 hdfs-site.xml，默认的文件系统就是 HDFS，否则就是本地文件系统)。
spark-shell默认是Hive支持的；代码中是默认不支持的，需要手动指定（加一个参数即可）。
3.4.1 使用内嵌Hive
如果使用 Spark 内嵌的 Hive, 则什么都不用做, 直接使用即可.
Hive 的元数据存储在 derby 中, 仓库地址:$SPARK_HOME/spark-warehouse
spark.sql("show tables").show
spark.sql("create table aa(id int)") # 建表
spark.sql("insert into aa values(100)") #插入数据
spark.sql("select * from aa").show #打印输出
3.4.2 使用外部Hive
如果Spark要接管Hive外部已经部署好的Hive，需要通过以下几个步骤。
确定原有Hive是正常工作的
需要把hive-site.xml拷贝到spark的conf/目录下
cp /opt/module/hive/conf/hive-site.xml /opt/module/spark-local/conf/
如果以前hive-site.xml文件中，配置过Tez相关信息，注释掉
把Mysql的驱动copy到Spark的jars/目录下
cp /opt/software/mysql-connector-java-5.1.48.jar /opt/module/spark-local/jars
需要提前启动hive服务，hive/bin/hiveservices.sh start
如果访问不到hdfs，则需把core-site.xml和hdfs-site.xml拷贝到conf/目录
再次执行查询表，可以看到查询的是外部的hive中的表
3.4.3 运行Spark SQL CLI
Spark SQLCLI可以很方便的在本地运行Hive元数据服务以及从命令行执行查询任务。在Spark目录下执行如下命令启动Spark SQ LCLI，直接执行SQL语句，类似Hive窗口。
bin/spark-sql
3.4.4 代码中操作Hive
1）添加依赖
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_2.11</artifactId>
<version>2.1.1</version>
</dependency>
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>1.2.1</version>
</dependency>
2）拷贝hive-site.xml到resources目录
3）代码测试
object SparkSQL02_Hive {
def main(args: Array[String]): Unit = {
//创建配置文件对象
val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL02_Hive")
//创建SparkSession对象(添加对Hive的支持)_
val spark: SparkSession = SparkSession.builder().enableHiveSupport().config(conf).getOrCreate()
spark.sql("show tables").show()
//释放资源
spark.stop()
}
}
输出：
+--------+-----------+-----------+
|database| tableName|isTemporary|
+--------+-----------+-----------+
| default| business| false|
| default| dept| false|
| default| emp| false|
| default| emp_sex| false|
| default| location| false|
| default| movie_info| false|
| default|person_info| false|
| default| score| false|
| default| stu| false|
| default| stu_buck| false|
| default| stu_par| false|
| default| student| false|
| default| video_ori| false|
+--------+-----------+-----------+
4. 项目实战
4.1 准备数据
我们这次Spark-sql操作所有的数据均来自 Hive，首先在Hive中创建表,并导入数据。一共有3张表： 1张用户行为表，1张城市表，1 张产品表。
建表语句：
CREATE TABLE `user_visit_action`(
`date` string,
`user_id` bigint,
`session_id` string,
`page_id` bigint,
`action_time` string,
`search_keyword` string,
`click_category_id` bigint,
`click_product_id` bigint,
`order_category_ids` string,
`order_product_ids` string,
`pay_category_ids` string,
`pay_product_ids` string,
`city_id` bigint)
row format delimited fields terminated by '\t';
load data local inpath '/opt/module/data/user_visit_action.txt' into table sparkpractice.user_visit_action;
CREATE TABLE `product_info`(
`product_id` bigint,
`product_name` string,
`extend_info` string)
row format delimited fields terminated by '\t';
load data local inpath '/opt/module/data/product_info.txt' into table sparkpractice.product_info;
CREATE TABLE `city_info`(
`city_id` bigint,
`city_name` string,
`area` string)
row format delimited fields terminated by '\t';
load data local inpath '/opt/module/data/city_info.txt' into table sparkpractice.city_info;
4.2.1 需求：各个区域热门商品Top3
这里的热门商品是从点击量的维度来看的，计算各个区域前三大热门商品，并备注上每个商品在主要城市中的分布比例，超过两个城市用其他显示。
例如：
4.2.2 思路分析
使用 sql 来完成，碰到复杂的需求，可以使用 udf 或 udaf
查询出来所有的点击记录，并与 city_info 表连接，得到每个城市所在的地区，与 Product_info 表连接得到产品名称
按照地区和商品名称分组，统计出每个商品在每个地区的总点击次数
每个地区内按照点击次数降序排列
只取前三名，并把结果保存在数据库中
城市备注需要自定义 UDAF 函数
4.2.3 代码实现
1）通过SQL实现求出各个地区热门排名在前3
1.1）从用户行为表中，查询所有点击记录，并和city_info,product_info进行连接
select
c.area,
p.product_name
from
user_visit_action a
join
city_info c
on
a.city_id = c.city_id
join
product_info p
on
a.click_product_id = p.product_id
where
a.click_product_id != -1
limit 10
1.2）按照地区和商品的名称进行分组，统计出每个地区每个商品的点击总数
select
t1.area,
t1.product_name,
count(*) as product_click_count
from
(
select
c.area,
p.product_name
from
user_visit_action a
join
city_info c
on
a.city_id = c.city_id
join
product_info p
on
a.click_product_id = p.product_id
where
a.click_product_id != -1
)t1
group by t1.area,t1.product_name
limit 10
1.3）针对每个地区，对商品点击数进行降序排序
select
t2.area,
t2.product_name,
t2.product_click_count,
row_number() over(partition by t2.area order by t2.product_click_count desc) cn
from
(
select
t1.area,
t1.product_name,
count(*) as product_click_count
from
(
select
c.area,
p.product_name
from
user_visit_action a
join
city_info c
on
a.city_id = c.city_id
join
product_info p
on
a.click_product_id = p.product_id
where
a.click_product_id != -1
)t1
group by t1.area,t1.product_name
)t2
limit 30
输出：
1.4）取当前地区的前3名
select
t3.area,
t3.product_name,
t3.product_click_count,
t3.cn
from
(
select
t2.area,
t2.product_name,
t2.product_click_count,
row_number() over(partition by t2.area order by t2.product_click_count desc) cn
from
(
select
t1.area,
t1.product_name,
count(*) as product_click_count
from
(
select
c.area,
p.product_name
from
user_visit_action a
join
city_info c
on
a.city_id = c.city_id
join
product_info p
on
a.click_product_id = p.product_id
where
a.click_product_id != -1
)t1
group by t1.area,t1.product_name
)t2
)t3
where t3.cn <= 3
limit 10
输出：
