文章目录
- 一、基本操作
- 二、默认数据源
- (一)默认数据源Parquet
- (二)案例演示读取Parquet文件
- 1、在Spark Shell中演示
- 练习1、将`student.txt`文件转换成`student.parquet`
- 练习2、读取`student.parquet`文件得到学生数据帧,并显示数据帧内容
- 2、在IntelliJ IDEA里演示
- 三、手动指定数据源
- (一)format()与option()方法概述
- (二)案例演示读取不同数据源
- 1、读取csv文件
- 2、读取json,保存为parquet
- 3、读取jdbc数据源,保存为json文件
- 四、数据写入模式
- (一)mode()方法
- (二)枚举类SaveMode
- (三)案例演示不同写入模式
- 五、分区自动推断
- (一)分区自动推断概述
- (二)分区自动推断演示
- 1、建四个文件
- 2、读取表数据
- 3、输出Schema信息
- 4、显示数据帧内容
- (三)分区自动推断注意事项
一、基本操作
Spark SQL提供了两个常用的加载数据和写入数据的方法:load()方法和save()方法。load()方法可以加载外部数据源为一个DataFrame,save()方法可以将一个DataFrame写入指定的数据源。
二、默认数据源
(一)默认数据源Parquet
默认情况下,load()方法和save()方法只支持Parquet格式的文件,Parquet文件是以二进制方式存储数据的,因此不可以直接读取,文件中包括该文件的实际数据和Schema信息,也可以在配置文件中通过参数spark.sql.sources.default对默认文件格式进行更改。Spark SQL可以很容易地读取Parquet文件并将其数据转为DataFrame数据集。
(二)案例演示读取Parquet文件
执行命令: cd $SPARK_HOME/examples/src/main/resources
,查看Spark的样例数据文件users.parquet
用cat命令显示users.parquet文件内容,只会显示乱码
启动hdfs:start-dfs.sh
将数据文件users.parquet上传到HDFS的/datasource/input目录
1、在Spark Shell中演示
启动spark服务:start-all.sh
启动Spark Shell,执行命令:spark-shell --master spark://master:7077
执行命令:val userdf = spark.read.load("hdfs://master:9000/datasource/input/users.parquet")
执行命令:userdf.show,查看数据帧内容
执行命令:userdf.printSchema,查看数据帧模式
执行命令:userdf.select("name", "favorite_color").write.save("hdfs://master:9000/datasource/output")
,对数据帧指定列进行查询,查询结果依然是数据帧,然后通过write成员的save()方法写入HDFS指定目录
查看HDFS上的输出结果
除了使用select()方法查询外,也可以使用SparkSession对象的sql()方法执行SQL语句进行查询,该方法的返回结果仍然是一个DataFrame。
基于数据帧创建临时视图,执行命令:userdf.createTempView("t_user")
执行SQL查询,将结果写入HDFS,执行命令:spark.sql("select name, favorite_color from t_user").write.save("hdfs://master:9000/datasource/output2")
查看HDFS上的输出结果
练习1、将student.txt
文件转换成student.parquet
解决思路:将student.txt转成studentDF,利用数据帧的save()方法保存到/datasource/output3目录,然后将文件更名复制到/datasource/input目录
得到学生数据帧 - studentDF
val ds = spark.read.textFile("hdfs://master:9000/student/input/student.txt")
case class Student(id: Int, name: String, gender: String, age: Int)
import spark.implicits._
val studentDS = ds.map(line => {
val fields = line.split(",")
val id = fields(0).toInt
val name = fields(1)
val gender = fields(2)
val age = fields(3).toInt
Student(id, name, gender, age)
}
)
val studentDF = studentDS.toDF()
studentDF.show
将学生数据帧保存为parquet文件,studentDF.write.save(“hdfs://master:9000/datasource/output3”)
查看生成的parquet文件
复制parquet文件到/datasource/input目录
练习2、读取student.parquet
文件得到学生数据帧,并显示数据帧内容
执行命令:val studentDF = spark.read.load(“hdfs://master:9000/datasource/input/student.parquet”)
执行命令:studentDF.show
2、在IntelliJ IDEA里演示
创建Maven项目
设置项目相关信息
单击【Finish】按钮
将java目录改成scala目录
在pom.xml文件里添加相关依赖,设置源程序文件夹
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>net.army.sql</groupId>
<artifactId>SparkSQLDemo</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.12.15</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.1.3</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.1.3</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
</build>
</project>
在resources目录里添加日志属性文件
log4j.rootLogger=ERROR, stdout, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spark.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
在resources目录里添加HFDS配置文件
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<property>
<description>only config in clients</description>
<name>dfs.client.use.datanode.hostname</name>
<value>true</value>
</property>
</configuration>
创建net.army.sql.day01包,在包里创建ReadParquetFile对象
package net.army.sql.day01
import org.apache.spark.sql.SparkSession
/**
* 功能:读取Parquet文件
* 作者:梁辰兴
* 日期:2023年06月12日
*/
object ReadParquetFile {
def main(args: Array[String]): Unit = {
// 创建或得到Spark会话对象
val spark = SparkSession.builder()
.appName("ReadParquetFile")
.master("local[*]")
.getOrCreate()
// 加载student.parquet文件,得到数据帧
val studentDF = spark.read.load("hdfs://master:9000/datasource/input/student.parquet")
// 显示学生数据帧内容
studentDF.show
// 查询20岁以上的女生
val girlDF = studentDF.filter("gender = '女' and age > 20")
// 显示女生数据帧内容
girlDF.show
// 保存查询结果到HDFS(保证输出目录不存在)
girlDF.write.save("hdfs://master:9000/datasource/output")
// 关闭Spark会话对象
spark.stop()
}
}
运行程序,查看控制台结果
三、手动指定数据源
(一)format()与option()方法概述
使用format()方法可以手动指定数据源。数据源需要使用完全限定名(例如org.apache.spark.sql.parquet),但对于Spark SQL的内置数据源,也可以使用它们的缩写名(JSON、Parquet、JDBC、ORC、Libsvm、CSV、Text)。
通过手动指定数据源,可以将DataFrame数据集保存为不同的文件格式或者在不同的文件格式之间转换。
在指定数据源的同时,可以使用option()方法向指定的数据源传递所需参数。例如,向JDBC数据源传递账号、密码等参数。
(二)案例演示读取不同数据源
1、读取csv文件
执行命令:cd $SPARK_HOME/examples/src/main/resources,查看Spark的样例数据文件people.csv
将people.csv文件上传到HDFS的/datasource/input目录,然后查看文件内
在Spark Shell里,执行命令:val peopleDF = spark.read.format(“csv”).load(“hdfs://master:9000/datasource/input/people.csv”),读取人员csv文件,得到人员数据帧
执行命令:peopleDF.show,查看人员数据帧内容
大家可以看到,people.csv文件第一行是字段名列表,但是转成数据帧之后,却成了第一条记录,这样显然是不合理的,怎么办呢?就需要用到option()方法来传递参数,告诉Spark第一行是表头header,而不是表记录。
执行命令:val peopleDF = spark.read.format(“csv”).option(“header”, “true”).load(“hdfs://master:9000/datasource/input/people.csv”)
执行命令:peopleDF.show,查看人员数据帧内容
由于csv文件默认分隔符是逗号,而people.csv的分隔符是分号,因此要利用option(“delimiter”, “;”)告诉Spark
执行命令:val peopleDF = spark.read.format(“csv”).option(“header”, “true”).option(“delimiter”, “;”).load(“hdfs://master:9000/datasource/input/people.csv”)
执行命令:peopleDF.show,查看人员数据帧内容
2、读取json,保存为parquet
查看people.json文件
将people.json上传到HDFS的/datasource/input目录,并查看其内容
在Spark Shell里,执行命令:val peopleDF = spark.read.format(“json”).load(“hdfs://master:9000/datasource/input/people.json”)
执行命令:peopleDF.show
执行命令:peopleDF.select(“name”, “age”).write.format(“parquet”).save(“hdfs://master:9000/datasource/output4”) (注意:format(“parquet”)其实可以省掉的)
查看生成的parquet文件(/datasource/output4/part-00000-a1e62c69-59e5-40b6-8391-89bdfffe61ff-c000.snappy.parquet)
将该parquet文件更名拷贝到/datasource/input目录,执行命令: hdfs dfs -cp /datasource/output4/part-00000-d0adfd21-9f55-49fc-a3dd-93bd313ea8e2-c000.snappy.parquet /datasource/input/people.parquet
现在读取/datasource/input/people.parquet文件得到人员数据帧
查看人员数据帧内容
3、读取jdbc数据源,保存为json文件
启动master的mysql服务
在Navicat创建mastermysql连接,连接到master虚拟机上安装的MySQL
查看student数据库里的user表
执行命令
val userDF = spark.read.format("jdbc")
.option("url", "jdbc:mysql://master:3306/student")
.option("driver", "com.mysql.jdbc.Driver")
.option("dbtable", "user")
.option("user", "root")
.option("password", "")
.load()
结果报错,没有找到数据库驱动程序
上传数据驱动程序到$SPARK_HOME/jars目录(每个节点都需要上传)
查看上传的数据库驱动程序
执行命令
val userDF = spark.read.format("jdbc")
.option("url", "jdbc:mysql://master:3306/student")
.option("driver", "com.mysql.jdbc.Driver")
.option("dbtable", "user")
.option("user", "root")
.option("password", "")
.load()
执行之后会有一个警告信息,通过设置useSSL=true来消除
执行命令
val userDF = spark.read.format("jdbc")
.option("url", "jdbc:mysql://master:3306/student?useSSL=false")
.option("driver", "com.mysql.jdbc.Driver")
.option("dbtable", "user")
.option("user", "root")
.option("password", "")
.load()
执行之后得到用户数据帧
执行命令:userDF.show,查看用户数据帧内容
执行命令:userDF.write.format(“json”).save(“hdfs://master:9000/datasource/output5”)
在虚拟机slave1查看生成的json文件,执行命令:hdfs dfs -cat /datasource/output5/*
四、数据写入模式
(一)mode()方法
在写入数据时,可以使用mode()方法指定如何处理已经存在的数据,该方法的参数是一个枚举类SaveMode。
使用SaveMode类,需要import org.apache.spark.sql.SaveMode;
(二)枚举类SaveMode
SaveMode.ErrorIfExists:默认值。当向数据源写入一个DataFrame时,如果数据已经存在,就会抛出异常。
SaveMode.Append:当向数据源写入一个DataFrame时,如果数据或表已经存在,会在原有的基础上进行追加。
SaveMode.Overwrite:当向数据源写入一个DataFrame时,如果数据或表已经存在,就会将其覆盖(包括数据或表的Schema)。
SaveMode.Ignore:当向数据源写入一个DataFrame时,如果数据或表已经存在,就不会写入内容,类似SQL中的CREATE TABLE IF NOT EXISTS。
(三)案例演示不同写入模式
查看数据源:people.json
查询该文件name里,采用覆盖模式写入/result,创建/result目录
执行命令:val peopledf = spark.read.format(“json”).load(“hdfs://master:9000/datasource/input/people.json”)
导入SaveMode类,执行命令:
import org.apache.spark.sql.SaveMode;
peopledf.select(“name”).write.mode(SaveMode.Overwrite).format(“json”).save(“hdfs://master:9000/result”)
在slave1虚拟机上查看生成的json文件
查询age列,以追加模式写入HDFS的/result目录,执行命令:peopledf.select(“age”).write.mode(SaveMode.Append).format(“json”).save(“hdfs://master:9000/result”)
在slave1虚拟机上查看追加生成的json文件
五、分区自动推断
(一)分区自动推断概述
表分区是Hive等系统中常用的优化查询效率的方法(Spark SQL的表分区与Hive的表分区类似)。在分区表中,数据通常存储在不同的分区目录中,分区目录通常以“分区列名=值”的格式进行命名。
以people作为表名,gender和country作为分区列,给出存储数据的目录结构
(二)分区自动推断演示
1、建四个文件
在master虚拟机上/home里创建如下目录及文件,其中目录people代表表名,gender和country代表分区列,people.json存储实际人口数据
{"name": "Wiek", "age": 85}
{"name": "Jenny", "age": 14}
{"name": "Alacn", "age": 35}
{"name": "张三丰", "age": 25}
{"name": "李诗琪", "age": 56}
{"name": "史蒂夫", "age": 38}
{"name": "王五", "age": 45}
{"name": "李四", "age": 44}
{"name": "张三", "age": 35}
{"name": "Alice", "age": 25}
{"name": "Mike", "age": 24}
{"name": "Linda", "age": 35}
2、读取表数据
执行命令:spark-shell,启动Spark Shell
执行命令:val peopledf = spark.read.format(“json”).load(“file:///home/people”)
3、输出Schema信息
执行命令:peopledf.printSchema()
4、显示数据帧内容
执行命令:peopledf.show()
从输出的Schema信息和表数据可以看出,Spark SQL在读取数据时,自动推断出了两个分区列gender和country,并将这两列的值添加到了数据帧peopledf中。
(三)分区自动推断注意事项
分区列的数据类型是自动推断的,目前支持数字、日期、时间戳、字符串数据类型。若不希望自动推断分区列的数据类型,则可以在配置文件中将spark.sql.sources.partitionColumnTypeInference.enabled
的值设置为false(默认为true,表示启用)。当禁用自动推断时,分区列将使用字符串
数据类型。