Spark从入门到精通26：Spark SQL使用数据源

1.什么是Parquet文件格式？

Spark SQL默认的数据源是Parquet格式，可以通过spark.sql.sources.default参数配置。

Parquet文件是一种列式存储格式的文件类型，具有以下特点：

• 可以跳过不符合条件的数据，只读取需要的数据，降低IO数据量
• 压缩编码可以降低磁盘存储空间。由于同一列的数据类型是一样的，可以使用更高效的压缩编码（例如Run Length Encoding和Delta Encoding等）进一步节约存储空间
• 只读取需要的列，支持向量运算，能够获取更好的扫描性能

2.使用通用函数（Load/Save）加载/保存数据

2.1使用load/save函数直接加载/保存数据

load/save函数默认的数据源是Parquet文件格式。

# cp $SPARK_HOME/examples/src/main/resources/* /root/input/

读取parquet文件：

scala> val usersDF = spark.read.load("file:///root/input/users.parquet")
usersDF: org.apache.spark.sql.DataFrame = [name: string, favorite_color: string ... 1 more field]

查看生成的DataFrame的表结构：

scala> usersDF.printSchema
root
|-- name: string (nullable = true)
|-- favorite_color: string (nullable = true)
|-- favorite_numbers: array (nullable = true)
| |-- element: integer (containsNull = true)

查看生成的DataFrame的数据：

scala> usersDF.show

image

查询用户的名字和喜欢的颜色，并保存：

scala> usersDF.select("name","favorite_color").write.save("file:///root/output/parquet")
scala> val usersDF = spark.read.load("file:///root/output/parquet")
usersDF: org.apache.spark.sql.DataFrame = [name: string, favorite_color: string]
scala> usersDF.show

image

parquet文件格式直接打开是乱码，可以重新加载到内存中查看结果文件。

2.2指定文件格式

Load/Save默认只能加载/保存parquet格式的文件，强行加载/保存其他格式的文件会出错。如需加载/保存其他格式的文件，需要调用format算子显示指定文件格式

scala> val peropleDF = spark.read.format("json").load("file:///root/input/people.json")
peropleDF: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
scala> peropleDF.show

image

2.3指定存储方式

可以使用SaveMode算子指定存储操作，SaveMode定义了对数据的处理模式。需要注意的是，这些保存模式不使用任何锁定，不是原子操作。SaveMode定义的存储方式有以下几种：

• SaveMode.ErrorIfExists：默认的存储方式，如果目标文件已存在，则抛出异常
• SaveMode.Append：如果目标文件已存在，则将新的内容追加到文件末尾
• SaveMode.Overwrite：如果目标文件已存在，则先清空原内容，再写入新的内容
• SavaMode.Ignore：如果目标文件已存在，则不执行任何操作

注意：当使用Overwrite方式写入时，在输出新数据之前，原来的数据就已经被删除了。

例如：直接使用下面的方式保存会出错，因为目标文件以及存在：

scala> usersDF.select($"name").write.save("file:///root/output/parquet")

image

设置为overwrite方式即可写入成功：

scala> usersDF.select($"name").write.mode("overwrite").save("file:///root/output/parquet")

2.4将结果保存为表

对查询到的结果集，可以使用saveAsTable保存为一张表：

scala> usersDF.select($"name").write.saveAsTable("table1")
scala> spark.sql("select * from table1").show

image

也可以使用partitionBy进行分区操作，使用bucketBy进行分桶操作等。

3.Parquet文件操作案例

Parquet是一个列格式而且用于多个数据处理系统中。Spark SQL提供支持杜宇Parquet文件的读写，也就是自动保存原始数据的schema。当写Parquet文件时，因为兼容性的缘故，所有的列被自动转化成nullable类型。

案例1：读入json格式的数据，将其转换成parquet格式，并创建相应的表来使用SQL进行查询。

scala> val empJSON = spark.read.json("file:///root/input/emp.json")
empJSON: org.apache.spark.sql.DataFrame = [comm: string, deptno: bigint ... 6 more fields]
scala> empJSON.write.mode("overwrite").parquet("file:///root/output/parquet")
scala> val empParquet = spark.read.parquet("file:///root/output/parquet")
empParquet: org.apache.spark.sql.DataFrame = [comm: string, deptno: bigint ... 6 more fields]
scala> empParquet.createOrReplaceTempView("tblEmp")
scala> spark.sql("select * from tblEmp where deptno = 10 and sal > 1500").show

image

案例2：Parquet Schema的合并。

Parquet支持Schema evolution（Schema 演变，即合并）。用户可以先定义一个简单的Schema，然后逐渐向Schema中增加列描述。通过这种方式，用户可以获取多个有不同Schema但是互相兼容的Parquet文件。

scala> val df1 = sc.makeRDD(1 to 5).map(i => (i, i2)).toDF("single","double")
df1: org.apache.spark.sql.DataFrame = [single: int, double: int]
scala> df1.write.parquet("file:///root/output/test_table/key=1")
scala> val df2 = sc.makeRDD(6 to 10).map(i => (i, i3)).toDF("single", "triple")
df2: org.apache.spark.sql.DataFrame = [single: int, triple: int]
scala> df2.write.parquet("file:///root/output/test_table/key=2")
scala> val df3 = spark.read.option("mergeSchema","true").parquet("file:///root/output/test_table")
df3: org.apache.spark.sql.DataFrame = [single: int, double: int ... 2 more fields]
scala> df3.printSchema()
root
|-- single: integer (nullable = true)
|-- double: integer (nullable = true)
|-- triple: integer (nullable = true)
|-- key: integer (nullable = true)
scala> df3.show

image

4.JSON文件操作案例

Spark SQL能自动解析JSON数据集的Schema，读取JSON数据集为DataFrame格式。读取JSON数据集方法为SQLContext.read().json()。该方法将String格式的RDD或JSON文件转换成DataFrame。需要注意的是，这里的JSON文件不是常规的JSON格式。JSON文件每一行必须包含一个独立的、自满足有效的JSON对象。如果用多行描述一个JSON对象，会导致读取出错。

案例：读取spark自带的示例文件people.json

scala> val path = "file:///root/input/people.json"
path: String = file:///root/input/people.json
scala> val peopleDF = spark.read.json(path)
peopleDF: org.apache.spark.sql.DataFrame = [age: bigint, name: string]
scala> peopleDF.printSchema
root
|-- age: long (nullable = true)
|-- name: string (nullable = true)
scala> peopleDF.show

image

scala> peopleDF.createOrReplaceTempView("tblPeople")
scala> spark.sql("select * from tblPeople where age = 19").show</pre>

image

5.JDBC 数据源操作案例

Spark SQL同样支持通过JDBC读取其他数据库的数据作为数据源。

案例：使用Spark SQL读取Oracle数据库中的表。

启动spark-shell的时候加载Oracle数据库驱动：

scala> spark-shell --master spark://localhost:7077 \
--jars /root/trainings/apache-hive-1.2.2-bin/lib/mysql-connector-java-5.1.46-bin.jar \
--driver-class-path /root/trainings/apache-hive-1.2.2-bin/lib/mysql-connector-java-5.1.46-bin.jar

方式1：

scala> val mysqlDF = spark.read.format("jdbc").
option("url","jdbc:mysql://localhost:3306/spark").
option("dbtable","tblUsers").
option("user","root").
option("password","123456").
load
mysqlDF: org.apache.spark.sql.DataFrame = [uID: int, uName: string ... 1 more field]

scala> mysqlDF.show

image

方式2：

使用java.util.Properties类封装属性：

scala> import java.util.Properties
import java.util.Properties
scala> val props = new Properties()
props: java.util.Properties = {}
scala> props.setProperty("user","root")
res1: Object = null
scala> props.setProperty("password","123456")
res2: Object = null

使用jdbc算子读取数据：

scala> val mysqlEmpDF = spark.read.jdbc(
"jdbc:mysql://localhost:3306/spark","tblUsers",props)
mysqlEmpDF: org.apache.spark.sql.DataFrame = [uID: int, uName: string ... 1 more field]

scala> mysqlEmpDF.show

image

6.Hive数据源操作案例

（1）配置Spark SQL支持Hive

将以下三个文件拷贝到$SPARK_HOME/conf目录下即可：

# cp $HIVE_HOME/conf/hive-site.xml $SPARK_HOME/conf
# cp $HADOOP_CONF_DIR/core-site.xml $SPARK_HOME/conf
# cp $HADOOP_CONF_DIR/hdfs-site.xml $SPARK_HOME/conf

（2）启动Spark shell的时候指定MySQL的驱动程序

# spark-shell --master spark://localhost:7077 \
--jars /root/trainings/apache-hive-1.2.2-bin/lib/mysql-connector-java-5.1.46-bin.jar \
--driver-class-path /root/trainings/apache-hive-1.2.2-bin/lib/mysql-connector-java-5.1.46-bin.jar

（3）使用spark-shell操作Hive

（*）查询Hive中有哪些表

scala> spark.sql("show tables").show

image

（*）在Hive中创建表

scala> spark.sql("create table tblStudents(sid int,sname string,age int)
row format delimited fields terminated by ',';")
18/07/09 00:00:05 WARN metastore.HiveMetaStore: Location: hdfs://localhost:9000/hive/warehouse/tblstudents specified for non-external table:tblstudents
res11: org.apache.spark.sql.DataFrame = []

（*）往Hive表中导入数据

scala> spark.sql("load data local inpath '/root/input/student.csv' into table tblstudents")
res12: org.apache.spark.sql.DataFrame = []

（*）查询Hive表中的数据

scala> spark.sql("select * from tblStudents").show

image

本节介绍了Spark SQL常用的几种数据源，更多数据源的使用请参考Spark的官方文档。