Spark读取parquet数据源

1.Parquet文件介绍

Apache Parquet是Hadoop生态圈中一种新型的列式存储格式&＃xff0c;它可以兼容Hadoop生态圈中大多数据的计算框架&＃xff0c;如Hadoop, Spark&＃xff0c;它也被多种查询引擎所支持&＃xff0c;例如Hive, Impala等&＃xff0c;而且它是跨语言和平台的。

Parquet的产生是由Twitter和Cloudera公司由于Apache Impala的缘故使用开发完成并开源给Apache基金会组织进行孵化&＃xff0c;现已成为APache的顶级项目。

另一方面&＃xff0c;随着嵌套格式数据的需求日益增加&＃xff0c;目前Hadoop生态圈中主流的OLAP都支持丰富的数据类型&＃xff0c;例如Hive, SparkSQL, Impala等都支持诸如array, map, struct这样的复合数据类型&＃xff0c;这也使得像Parquet这种原生支持嵌套数据的存储格式变得至关重要&＃xff0c;由于它是列式存储&＃xff0c;所以在性能方面会很高。

列式存储&＃xff0c;就是按照列进行存储数据&＃xff0c;把某一旬的数据连续地存储&＃xff0c;每一行中的不同的列离散分布。相比较于行存储&＃xff0c;列存储具有以下优势&＃xff1a;

• 可以跳过不符合条件的数据&＃xff0c;只读取需要的数据&＃xff0c;降低磁盘IO
• 使用压缩可以降低磁盘的存储空间&＃xff0c;并且由于同一列的数据类型是一样的&＃xff0c;可以使用更高效的压缩编码进一步节约存储空间&＃xff1b;
• 只读取需要的列&＃xff0c;能够获得更好的扫描性能&＃xff1b;

Parquet是SparkSQL默认的存储格式&＃xff0c;它支持灵活的读写Parquet文件&＃xff0c;并对Parquet文件的schema可以自动解析。

import org.apache.spark.sql.{DataFrame, SparkSession}object SparkSqlParquetSource { def main(args: Array[String]): Unit &＃61; { val spark: SparkSession &＃61; SparkSession.builder() .appName(SparkSqlParquetSource.getClass.getSimpleName) .master("local") .getOrCreate() spark.sparkContext.setLogLevel("WARN") //读取json文件生成DataFrame val sanguoDF: DataFrame &＃61; spark.read.format("json").load("./dataset/sanguo.json") //把结果写入parquet sanguoDF.write.parquet("./dataset/parquet/sanguo.parquet") /** * 读取刚刚写入的parquet文件 * */ val sgDF: DataFrame &＃61; spark.read.parquet("./dataset/parquet/sanguo.parquet") //打印schema sgDF.printSchema() sgDF.show() //释放资源 spark.stop() }}

上述代码中&＃xff0c;通过读取json文件写入parquet文件&＃xff1a;

.crc是校验文件&＃xff0c;数据生成的parquet文件是用snappy压缩的

运行的结果

2.分区发现(Partition Discovery)

表分区是一种常见的优化方法&＃xff0c;比如Hive中就提供了分区表的特性。在一个分区表中&＃xff0c;不同分区的数据通常是存储在HDFS上不同的目录中&＃xff0c;分区列的值通常就包含在了分区目录的目录名中。SparkSQL中的parquet数据源&＃xff0c;支持自动根据目录名推断出分区信息。例如&＃xff0c;如果将人口数据存储在分区表中&＃xff0c;并且使用性别和国家作为分区列。那么目录结构可能如下所示&＃xff1a;

path└── to └── table ├── gender&＃61;male │ ├── ... │ │ │ ├── country&＃61;US │ │ └── data.parquet │ ├── country&＃61;CN │ │ └── data.parquet │ └── ... └── gender&＃61;female ├── ... │ ├── country&＃61;US │ └── data.parquet ├── country&＃61;CN │ └── data.parquet └── ...

如果将 path/to/table传入SparkSession.read.parquet()或SparkSession.read.load()方法&＃xff0c;那么SparkSQL就会自动根据目录的结构&＃xff0c;推断出分区信息是gender和country。即使数据文件中只包含了两列值&＃xff0c;name和age&＃xff0c;但是Spark SQL返回的DataFrame&＃xff0c;调用printSchema()方法时&＃xff0c;会打印出四个列的值&＃xff1a;name&＃xff0c;age&＃xff0c;country&＃xff0c;gender。这就是自动分区推断的功能。

此外&＃xff0c;分区列的数据类型&＃xff0c;也是自动被推断出来的。目前&＃xff0c;Spark SQL仅支持自动推断出数字类型和字符串类型。有时&＃xff0c;用户也许不希望Spark SQL自动推断分区列的数据类型。此时只要设置一个配置即可&＃xff0c; spark.sql.sources.partitionColumnTypeInference.enabled&＃xff0c;默认为true&＃xff0c;即自动推断分区列的类型&＃xff0c;设置为false&＃xff0c;即不会自动推断类型。禁止自动推断分区列的类型时&＃xff0c;所有分区列的类型&＃xff0c;就统一默认都是String。

3.元数据合并(Schema Merging)

如同ProtocolBuffer&＃xff0c;Avro&＃xff0c;Thrift一样&＃xff0c;Parquet也是支持元数据合并的。用户可以在一开始就定义一个简单的元数据&＃xff0c;然后随着业务需要&＃xff0c;逐渐往元数据中添加更多的列。在这种情况下&＃xff0c;用户可能会创建多个Parquet文件&＃xff0c;有着多个不同的但是却互相兼容的元数据。Parquet数据源支持自动推断出这种情况&＃xff0c;并且进行多个Parquet文件的元数据的合并。

因为元数据合并是一种相对耗时的操作&＃xff0c;而且在大多数情况下不是一种必要的特性&＃xff0c;从Spark 1.5.0版本开始&＃xff0c;默认是关闭Parquet文件的自动合并元数据的特性的。可以通过以下两种方式开启Parquet数据源的自动合并元数据的特性&＃xff1a;

1. 读取Parquet文件时&＃xff0c;将数据源的选项&＃xff0c;mergeSchema&＃xff0c;设置为true
2. 将spark.sql.parquet.mergeSchema参数设置为true

import org.apache.spark.sql.{DataFrame, Dataset, SaveMode, SparkSession}object SparkSqlSchemaMergeTest { def main(args: Array[String]): Unit &＃61; { val spark: SparkSession&＃61; SparkSession.builder() .master("local") .appName(SparkSqlSchemaMergeTest.getClass.getSimpleName) .getOrCreate() spark.sparkContext.setLogLevel("WARN") //导入隐式转换 import spark.implicits._ //创建第一个DataFrame val personSeq: Seq[(String, Int)] &＃61; Array(("风清扬", 55), ("任我行", 60)).toSeq val personDF: DataFrame &＃61; spark.createDataset(personSeq).toDF("name", "age")// personDF.printSchema()// personDF.show() //保存第一个DF到parquet文件 personDF.write.mode(SaveMode.Append).parquet("./dataset/parquet/person.parquet") //创建第二个DataFrame val personWithGenderSeq: Seq[(String, String)] &＃61; Array(("关羽", "男"), ("张飞", "男")).toSeq val personWithGenderDF: DataFrame &＃61; spark.createDataset(personWithGenderSeq).toDF("name", "gender") //保存第二个DF到parquet文件 personWithGenderDF.write.mode(SaveMode.Append).parquet("./dataset/parquet/person.parquet") /** * 首先&＃xff0c;第一个DataFrame和第二个DataFrame的元数据是肯定不一样的 * 一个是包含了name和age, 而别一个是包含了name和gender * 所以&＃xff0c;期望将来读取这个parquet文件时只有三列name, age, gender, 实现自动合并元数据的功能 * */ //用mergeSchema的方式&＃xff0c;读取person.parquet文件中的数据&＃xff0c;并将元数据合并 val mergedPerosnDF: DataFrame &＃61; spark.read.format("parquet").option("mergeSchema", "true").load("./dataset/parquet/person.parquet") //打印合并的schema的信息 mergedPerosnDF.printSchema() mergedPerosnDF.show() spark.stop() }}

代码实现截图

运行结果可以看出进行了schema的合并