Spark SQL下的Parquet使用最佳实践和代码实战

Posted 2020-08-28

一：Spark SQL下的Parquet使用最佳实践

1，过去整个业界对大数据的分析的技术栈的Pipeline一般分为一下两种方式：

A）Data Source -> HDFS -> MR/Hive/Spark(相当于ETL) -> HDFS Parquet -> SparkSQL/impala -> Result Service(可以放在DB中，也有可能被通过JDBC/ODBC来作为数据服务使用)；

B）Data Source -> Real time update data to Hbase/DB -> Export to Parquet -> SparkSQL/impala -> Result Service(可以放在DB中，也有可能被通过JDBC/ODBC来作为数据服务使用)；

上述的第二种方式完全可以通过Kafka+Spark Streaming+Spark SQL（内部也强烈建议采用Parquet的方式来存储数据）的方式取代。

2，期待的方式：Data Source -> Kafka -> Spark Streaming -> Parquet -> Spark SQL（ML、Graphx等）-> Parquet -> 其他各种Data Mining等。

二：Parquet的精要介绍

摘自官网：

Apache Parquet is a columnar storage format available to any project in the Hadoop ecosystem, regardless of the choice of data processing framework, data model or programming language.

1，Parquet是列式存储格式的一种文件类型，列式存储有以下的核心：

A.可以跳过不符合条件的数据，只读取需要的数据，降低IO数据量。

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

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

三：Spark SQL下的Parquet意义再思考

1，如果说HDFS是大数据时代文件系统的事实标准的话，Parquet就是大数据时代存储格式的事实标准；

2，速度更快：从使用Spark SQL操作普通文件CSV和Parquet文件的速度对比上来看，绝大多数情况下使用Parquet会比使用CSV等普通文件速度提升10倍左右；（在一些普通文件系统无法再Spark上成功运行程序的情况下，使用Parquet很多时候都可以成功运行）；

3，Parquet的压缩技术非常稳定出色，在Spark SQL中对压缩技术的处理可能无法正常的完成工作（例如会导致Lost Task，Lost Exexutor），但是此时如果使用Parquet就可以正常的完成；

4，极大的减少磁盘I/O，通常情况下能够减少75%的存储空间，由此可以极大地减少Spark SQL处理数据的时候的数据输入内容，尤其是在Spark 1.6.x中下推过滤器在一些情况下可以极大的进一步减少磁盘的I/O和内存的占用；

5，Spark 1.6.x+Parquet极大的提升了数据扫描的吞吐量，这极大的提高了数据的查找速度，Spark 1.6和Spark 1.5相比较而言提升了1倍的速度，在Spark 1.6.x中操作Parquet时候CPU的使用也进行了极大的优化，有效的降低了CPU的使用；

6，采用Parquet可以极大的优化Spark的调度和执行，我们测试表面Spark如果采用Parquet可以有效的减少Stage的执行消耗，同时可以优化执行路径；

四：Spark SQL下的Parquet内幕解密

1，列式存储是以什么基本格式来存储数据的？表现上是树状结构，在内部有元数据的Table；

2，在具体的Parquet文件存储的时候有三个组成部分：

A）Storage Format：Parquet定义了具体的数据内部的类型和存储格式；

B）Object Model Converters：Parquet中负责计算框架中数据对象和Parquet文件中具体数据类型的映射；

C）Object Models：在Parquet中具有自己的Object Model定义的存储格式，例如说Avro具有自己的Object Model，但是Parquet在处理相关的格式的数据的时候使用自己的Object Model来存储；

映射完成后Parquet会进行自己的Column Encoding，然后存储成为Parquet格式的文件

3，Modules

The parquet-format project contains format specifications and Thrift definitions of metadata required to properly read Parquet files.

The parquet-mr project contains multiple sub-modules, which implement the core components of reading and writing a nested, column-oriented data stream, map this core onto the parquet format, and provide Hadoop Input/Output Formats, Pig loaders, and other Java-based utilities for interacting with Parquet.

The parquet-compatibility project contains compatibility tests that can be used to verify that implementations in different languages can read and write each other’s files.

4，举例说明：

message AddressBook {

 required string owner;

 repeated string ownerPhoneNumbers;

 repeated group contacts {

   required string name;

   optional string phoneNumber;

 }

}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17

required（出现1次），optional（出现0次或者1次），repeated（出现0次或者多次）

这个schema中每条记录表示一个人的AddressBook。有且只有一个owner，owner可以有0个或者多个ownerPhoneNumbers，owner可以有0个或者多个contacts。每个contact有且只有一个name，这个contact的phoneNumber可有可无。

第一点：就存储数据本身而言，只考虑叶子节点，我们的叶子节点owner、ownerPhoneNumber、name、phoneNumber；

第二点：在逻辑上而言Schema实质上是一个Table：

AddressBook
owner	ownerphonenumber	contacts
		name	phonenumber

第三点：对于一个Parquet文件而言，数据会被分成Row Group（里面包含很多Column，每个Column具有几个非常重要的特性例如Repetition Level、Definition Level）；

第四点：Column在Parquet中是以Page的方式存在的，Page中有Repetition Level、Definition Level等内容；

第五点：Row Group在Parquet中是数据读写的缓存单元，所以对Row Group的设置会极大的影响Parquet的使用速度和效率，所以如果是分析日志的话，我们一般建议把Row Group的缓存大小配置成大约256MB，很多人的配置都是大约1G，如果想最大化的运行效率强烈建议HDFS的Block大小和Row Group一致；

第六点：在实际存储的把一个树状结构，通过巧妙的编码算法，转换成二维表结构

Repetition Level	Definition Level	Value
1	2	132990600
0	1	“spark”
0	0	NULL