第61课：Spark SQL数据加载和保存内幕深度解密实战

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

本期内容：

1、Spark SQL加载数据

2、Spark SQL保存数据

3、Spark SQL对数据处理的思考

Spark SQL对数据的操作涉及到对数据的输入、输出，其实主要就是操作DataFrame。通过DataFrameReader的load方法可以创建DataFrame，通过DataFrameWriter的save方法可以把DataFrame中的数据保存到具体文件。我们可以通过设置具体的格式参数来指出读取的文件是什么类型，以及通过设置具体的格式参数来指出输出的文件是什么类型。

一、Spark SQL加载数据

SQLContext中有load方法，实际上也就是使用了DataFrameReader的load方法。目前的Spark 1.6.0中,虽然SQLContext还有load方法，但已被设置为@deprecated。也就是说在Spark 2.0中SQLContext可能不再使用load方法。

DataFrameReader用来从外部存储系统（比如：文件系统，键值存储库，等等）读取数据，并返回DataFrame对象。

DataFrameReader有format方法，用来设置输入数据源的格式。

DataFrameReader用load方法，把外部数据加载进来，变成一个DataFrame。各种load方法，最终实际是使用以下load方法：

DataFrameReader的load方法是生成ResolvedDataSource对象，再使用该对象生成并返回一个DataFrame对象。

查看ResolvedDataSource的apply方法，可以发现其中是通过模式匹配来对各种类型的数据源分别进行处理，生成ResolvedDataSource对象。

查看ResolvedDataSource的backwardCompatibilityMap，可以知道，Spark SQL可以用来处理jdbc访问的数据库、json文件、parquet等多种来源的数据。

查看ResolvedDataSource的lookupDataSource方法，可以知道，Spark SQL可以处理Avro、Hive的ORC、访问Hadoop的文件系统等各种文件。

理论上讲，Spark SQL可以支持所有格式数据源。

DataFrame类中有this、toDF、printSchema、explain、show等常用方法。

二、Spark SQL保存数据

DataFrame中有各种save方法，实际上也就是使用了DataFrameWriter的save方法。

目前的Spark 1.6.0中虽然DataFrame还有save方法，但已被设置为@deprecated。也就是说在Spark 2.0中DataFrame可能不再使用save方法。用以下程序SparkSQLLoadSaveOps.Java来演示Spark SQL 对数据的读取和保存：

1. package com.dt.spark.SparkApps.sql;
   
2. 
   
3. import org.apache.spark.SparkConf;
   
4. import org.apache.spark.api.java.JavaSparkContext;
   
5. import org.apache.spark.sql.DataFrame;
   
6. import org.apache.spark.sql.SQLContext;
   
7. 
   
8. /**
   
9. * 用SparkSQL读取和保存数据。
   
10. * @author DT大数据梦工厂
    
11. * 新浪微博：http://weibo.com/ilovepains/
    
12. */
    
13. public class SparkSQLLoadSaveOps {
    
14.         public static void main(String[] args) {
    
15.                 SparkConf conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrameByProgrammatically");
    
16.                 JavaSparkContext sc = new JavaSparkContext(conf);
    
17.                 SQLContext sqlContext = new SQLContext(sc);
    
18.                
    
19.                 DataFrame peopleDF = sqlContext.read().format("json").load("D:\\Big_Data_Software\\spark-1.6.0-bin-hadoop2.6\\examples\\src\\main\\resources\\people.json");
    
20.                 peopleDF.select("name").write().format("json").save("D:\\Big_Data_Software\\spark-1.6.0-bin-hadoop2.6\\examples\\src\\main\\resources\\peopleNames.json");
    
21.         }
    
22. }

复制代码

运行以上程序，可以发现会有相应的路径有文件输出：

我们把代码中最后两行的format方法去掉，并修改源数据和目标数据路径：

1. DataFrame peopleDF = sqlContext.read().load("D:\\Big_Data_Software\\spark-1.6.0-bin-hadoop2.6\\examples\\src\\main\\resources\\users.parquet");
   
2.                 peopleDF.select("name").write().save("D:\\Big_Data_Software\\spark-1.6.0-bin-hadoop2.6\\examples\\src\\main\\resources\\usersNames");

复制代码

保存并运行程序，再看输出结果：

我们可以看到其中有Parquet文件。可见Spark SQL的缺省输出格式是Parquet。 我们仍然修改代码中最后两行，源数据使用json文件，输出时只增加mode方法以修改保存模式为追加：

1. DataFrame peopleDF = sqlContext.read().format("json").load("D:\\Big_Data_Software\\spark-1.6.0-bin-hadoop2.6\\examples\\src\\main\\resources\\people.json");
   
2.         peopleDF.select("name").write().mode(SaveMode.Append).save("D:\\Big_Data_Software\\spark-1.6.0-bin-hadoop2.6\\examples\\src\\main\\resources\\usersNames");

复制代码

保存并运行程序，再看输出结果：

可以发现追加了parquet文件。

三、Spark SQL对数据处理的思考

业界的主流公司在做大数据分析的时候，基本上都以Parquet方式存储数据。Parquet文件是基于列式存储的，列之间是分开的，可以进行各种高效的优化。Spark底层一般都会接 Parquet文件。Cloudera开发出了Kudu，Kudu是列式存储引擎，感觉似乎不错，但还不够稳定。Hadoop的HDFS有被替代的可能。我们再来探究一下缺省输出格式是Parquet的原因。

先看一下DataFrameWriter.scala源码：

conf是SQLConf对象。看SQLConf. defaultDataSourceName的定义：

再看DEFAULT_DATA_SOURCE_NAME的定义：

可以看到默认方式下确实是Parquet方式。

注：本学习笔记来自DT大数据梦工厂        微信公众号：DT_Spark        每晚8点YY永久直播频道：68917580

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：Spark Park SPAR SPA sql Spark scala DT_Spark 大数据