Spark SQL数据加载、存储概述

Spark SQL的默认数据源格式为Parquet格式。当数据源为Parquet文件时，Spark SQL可以方便地进行读取，甚至可以直接在Parquet文件上执行查询操作。修改配置项spark.sql.sources.default，可以修改默认数据源格式。

以下示例通过通用的load\save方法对Parquet文件进行读取和存储

val usersDF = sparkSession.read.load("examples/src/main/resources/users.parquet")

usersDF.select("name", "favorite_color").write.save ("namesAndFavColors.parquet")

正如前面所讲，sparkSession是Spark SQL的编程主入口，在读取数据源时，需要调用sparkSession.read方法返回一个DataFrameReader对象，进而通过其提供的、读取各种结构化数据源的方法来读取数据源，其中包括通用的load方法，返回的是DataFrame对象。

同样地，在上例第2行通过DataFrame.write方法返回了一个DataFrameWriter对象，进而调用其通用save方法，将DataFrame对象以Parquet文件格式存储。

Parquet是面向分析型业务的列式存储格式，由Twitter和Cloudera合作开发，2015年5月从Apache的孵化器里毕业成为Apache顶级项目。

Parquet是典型的列式存储格式，和行式存储相比具有以下优势：

（1）可以跳过不符合条件的数据，只读取需要的数据，从而降低I/O数据量。

（2）压缩编码可以降低磁盘存储空间。由于同一列的数据类型是一样的，因此可以使用更高效的压缩编码（例如Run Length Encoding和Delta Encoding）进一步节约存储空间。

（3）只读取需要的列，支持向量运算，能够获取更好的扫描性能。

之所以开发Parquet，是因为当时Twitter的日增数据量压缩之后达到100TB+，并存储在HDFS上，工程师会使用多种计算框架（例如MapReduce、Hive、Pig等）对这些数据进行分析和挖掘；日志结构是复杂的嵌套数据类型，例如一个典型的日志的schema有87列，嵌套了7层。因此需要设计一种列式存储格式，既能支持关系型数据（简单数据类型），又能支持复杂的嵌套类型的数据，同时能够适配多种数据处理框架。

关系型数据的列式存储可以将每一列的值直接排列下来，不用引入其他的概念，也不会丢失数据。关系型数据的列式存储比较好理解，而理解嵌套类型数据的列式存储则会遇到一些麻烦。如图6-1所示，我们把嵌套数据类型的一行叫作一个记录（record)。嵌套数据类型的特点是，对于一个record中的column，除了可以是Int、Long、String这样的原语（primitive）类型以外，还可以是List、Map、Set这样的复杂类型。在行式存储中一行的多列是连续写在一起的，在列式存储中数据按列分开存储，例如可以只读取A.B.C这一列的数据而不去读A.E和A.B.D。那么如何根据读取出来的各个列的数据重构出一行记录呢？

图6-1  行式存储和列式存储

Google的Dremel系统解决了这个问题，核心思想是使用“record shredding and assembly algorithm”（记录粉碎与组装算法）来表示复杂的嵌套数据类型，同时辅以按列的高效压缩和编码技术，从而降低存储空间，提高I/O效率，降低上层应用延迟。Parquet就是基于Dremel的数据模型和算法实现的。

当数据源不是Parquet格式文件时，需要手动指定数据源的格式。数据源格式需指定全名（如org.apache.spark.sql.parquet）；如果数据源为内置格式，则只需指定简称（json、parquet、jdbc、orc、libsvm、csv、text）即可。通过指定数据源格式名，还可以对DataFrame进行类型转换操作。

以下示例是将原为JSON格式的数据源转储为Parquet格式文件：

val peopleDF = spark.read.format("json").load("examples/src/main/resources /people.json")

peopleDF.select("name", "age").write.format("parquet").save("namesAndAges.parquet")

相比于使用read API将文件加载到DataFrame并对其进行查询，还可以使用SQL直接查询该文件。示例如下：

val sqlDF = spark.sql("SELECT * FROM  parquet.`examples/src/main/resources /users.parquet`")

需要注意的是，在使用SQL直接查询Parquet文件时，需加“parquet.”标识符和Parquet文件所在路径。

保存操作可以选择使用存储模式（SaveMode），从而指定如何处理现有数据（如果存在），如表6-1所示。例如将数据追加到文件或者是覆盖文件内容。需要注意的是，这些存储模式不会使用任何锁定，也不是原子的。另外，当执行覆盖时，在写入新数据之前，旧数据将被删除。

以下是存储模式运用示例，通过mode()方法设置将数据写入指定文件的存储模式：

people2DF.select("name", "age").write().mode(SaveMode.Append).save (“hdfs://hadoop1:9000/output/namesAndAges.parquet”);

DataFrame也可以使用saveAsTable()方法作为持久化表保存到Hive metastore（Hive元数据库）中。需要注意的是，即使用户没有在集群中部署现有的Hive数据仓库以供持久化表的存储，也可以使用该功能，因为Spark将创建默认的本地Hive metastore（使用Derby）。与createOrReplaceTempView()方法不同，saveAsTable()将实现DataFrame的内容，并创建一个指向Hive metastore中指定持久化表的数据的指针。只要保持与同一个metastore的连接，Spark程序重新启动后，持久化表也仍然存在。可以通过调用SparkSession上的table（"table_name"）方法通过指定持久化表的表名来重新创建持久化表对应的DataFrame对象。

对于基于文件的数据源，例如文本文件、Parquet文件、JSON文件等，也可以通过路径选项自定义指定表的存储路径，例如peopleDF.write.option.("path","examples/src/main/resources/ people.parquet").saveAsTable("people"）。当表被删除时，自定义表路径不会被删除，并且表数据仍然存在。如果未指定自定义表路径，Spark会将数据写入Hive仓库目录下的默认表路径。当表被删除时，默认的表路径也将被删除。

从Spark 2.1开始，持久性数据源表将表的每个分区的元数据分开存储在Hive metastore中。这带来了以下好处：

• 由于metastore可以仅返回查询涉及的必要的分区数据，因此不必再为每一个查询遍历查询表的所有数据。
• 涉及表的分区的Hive DDL语句，如ALTER TABLE PARTITION ... SET LOCATION，现在可用于使用Datasource API创建的表。

注意，在创建外部数据源表（带有路径选项的表）时，默认情况下不会收集分区信息。要同步转移中的分区信息，可以调用MSCK重新建立表的分区信息。

对于基于文件的数据源，可以根据需求对输出进行桶操作（bucket）。桶操作是一种将表或分区中的指定列的值作为key进行Hash分配到预定义数量的桶中的方法。这种操作有助于数据的均匀分布，从而支持高效的数据采样和查询。例如map、join、sort和partion操作。bucket和sort仅适用于持久化表。

对持久化表按name列进行桶操作，并指定生成42个桶（容器），且按age对数据排序：

peopleDF.write.bucketBy(42, "name").sortBy("age").saveAsTable("people_bucketed")

将DataFrame对象存储为按照favorite_color列值分区的Parquet文件：

usersDF.write.partitionBy("favorite_color").format("parquet").save("namesPartByColor.parquet")

对持久化表组合进行分区操作和桶操作：

peopleDF

  .write

  .partitionBy("favorite_color")

  .bucketBy(42, "name")

  .saveAsTable("people_partitioned_bucketed")

Spark SQL如同关系数据库一样，能够支持数据表的bucket（Hive桶操作）、排序和分区操作，这些操作可以根据业务需求对表内数据进行管理存储，从而提升查询效率。至于怎样根据实际的业务需求，对表合理地进行排序、分区操作，读者可以进一步查阅数据库相关书籍中的数据的排序和分区等内容。