Spark Parquet详解

Spark - Parquet

概述

Apache Parquet属于Hadoop生态圈的一种新型列式存储格式，既然属于Hadoop生态圈，所以也兼容大多圈内计算框架（Hadoop、Spark），另外Parquet是平台、语言无关的，这使得它的适用性很广，只要相关语言有对应支持的类库就能够用；

Parquet的优劣对比：

• 支持嵌套结构，这点对比一样是列式存储的OCR具有必定优点；
• 适用于OLAP场景，对比CSV等行式存储结构，列示存储支持映射下推和谓词下推，减小磁盘IO；
• 一样的压缩方式下，列式存储由于每一列都是同构的，所以可使用更高效的压缩方法；

下面主要介绍Parquet如何实现自身的相关优点，毫不仅仅是使用了列式存储就完了，而是在数据模型、存储格式、架构设计等方面都有突破；

列式存储 vs 行式存储

区别在于数据在内存中是以行为顺序存储仍是列为顺序，首先没有哪一种方式更优，主要考虑实际业务场景下的数据量、经常使用操做等；

数据压缩

例如两个学生对象分别在行式和列式下的存储状况，假设学生对象具有姓名-string、年龄-int、平均分-double等信息：

行式存储：

姓名	年龄	平均分	姓名	年龄	平均分
张三	15	82.5	李四	16	77.0

列式存储：

姓名	姓名	年龄	年龄	平均分	平均分
张三	李四	15	16	82.5	77.0

乍一看彷佛没有什么区别，事实上如何不进行压缩的化，两种存储方式实际存储的数据量都是一致的，那么确实没有区别，可是实际上如今经常使用的数据存储方式都有进行不一样程度的压缩，下面咱们考虑灵活进行压缩的状况下两者的差别：

行式存储是按照行来划分最小单元，也就是说压缩对象是某一行的数据，此处就是针对（张3、1五、82.5）这个数据组进行压缩，问题是该组中数据格式并不一致且占用内存空间大小不一样，也就无法进行特定的压缩手段；

列式存储则不一样，它的存储单元是某一列数据，好比（张3、李四）或者（15，16），那么就能够针对某一列进行特定的压缩，好比对于姓名列，假设咱们值到最长的姓名长度那么就能够针对性进行压缩，一样对于年龄列，通常最大不超过120，那么就可使用tiny int来进行压缩等等，此处利用的就是列式存储的同构性；

注意：此处的压缩指的不是相似gzip这种通用的压缩手段，事实上任何一种格式均可以进行gzip压缩，这里讨论的压缩是在此以外可以进一步针对存储数据应用更加高效的压缩算法以减小IO操做；

谓词下推

与上述数据压缩相似，谓词下推也是列式存储特有的优点之一，继续使用上面的例子：

行式存储：

姓名	年龄	平均分	姓名	年龄	平均分
张三	15	82.5	李四	16	77.0

列式存储：

姓名	姓名	年龄	年龄	平均分	平均分
张三	李四	15	16	82.5	77.0

假设上述数据中每一个数据值占用空间大小都是1，所以两者在未压缩下占用都是6；

咱们有在大规模数据进行以下的查询语句：

SELECT 姓名,年龄 FROM info WHERE 年龄>=16;

这是一个很常见的根据某个过滤条件查询某个表中的某些列，下面咱们考虑该查询分别在行式和列式存储下的执行过程：

• 行式存储：
  • 查询结果和过滤中使用到了姓名、年龄，针对所有数据；
  • 因为行式是按行存储，而此处是针对所有数据行的查询，所以须要遍历全部数据并对比其年龄数据，肯定是否返回姓名、年龄；
• 列式存储：
  • 过滤中使用了年龄，所以把年龄列取出来进行判断，判断结果是李四知足要求；
  • 按照上述判断结果把姓名列取出来，取出其中对应位置的姓名数据，与上述年龄数据一块儿返回；
  • 能够看到此时因为未涉及平均分，所以平均分列没有被操做过；

事实上谓词下推的使用主要依赖于在大规模数据处理分析的场景中，针对数据中某些列作过滤、计算、查询的状况确实更多，这一点有相关经验的同窗应该感触不少，所以这里只能说列式存储更加适用于该场景；

统计信息

这部分直接用例子来理解，仍是上面的例子都是有一点点改动，为了支持一些频繁的统计信息查询，针对年龄列增长了最大和最小两个统计信息，这样若是用户查询年龄列的最大最小值就不须要计算，直接返回便可，存储格式以下：

行式存储：

姓名	年龄	平均分	姓名	年龄	平均分	年龄最大	年龄最小
张三	15	82.5	李四	16	77.0	16	15

列式存储：

姓名	姓名	年龄	年龄	年龄最大	年龄最小	平均分	平均分
张三	李四	15	16	16	15	82.5	77.0

在统计信息存放位置上，因为统计信息一般是针对某一列的，所以列式存储直接放到对应列的最后方或者最前方便可，行式存储须要单独存放；

针对统计信息的耗时主要体如今数据插入删除时的维护更新上：

• 行式存储：插入删除每条数据都须要将年龄与最大最小值进行比较并判断是否须要更新，若是是插入数据，那么更新只须要分别于最大最小进行对比便可，若是是删除数据，那么若是删除的偏偏是最大最小值，就还须要从现有数据中遍历查找最大最小值来，这就须要遍历全部数据；
• 列式存储：插入有统计信息的对应列时才须要进行比较，此处若是是插入姓名列，那就没有比较的必要，只有年龄列会进行此操做，一样对于年龄列进行删除操做后的更新时，只须要针对该列进行遍历便可，这在数据维度很大的状况下能够缩小N（N为数据列数）倍的查询范围；

数据架构

这部分主要分析Parquet使用的数据模型，以及其如何对嵌套类型的支持（须要分析repetition level和definition level）；

数据模型这部分主要分析的是列式存储如何处理不一样行不一样列之间存储上的歧义问题，假设上述例子中增长一个兴趣列，该列对应行能够没有数据，也能够有多个数据（也就是说对于张三和李四，能够没有任何兴趣，也能够有多个，这种状况对于行式存储不是问题，可是对于列式存储存在一个数据对应关系的歧义问题），假设兴趣列存储以下：

兴趣	兴趣
羽毛球	篮球

事实上咱们并不肯定羽毛球和篮球到底都是张三的、都是李四的、仍是二人一人一个，这是由兴趣列的特殊性决定的，这在Parquet数据模型中称这一列为repeated的；

数据模型

上述例子的数据格式用parquet来描述以下：

message Student{
    required string name;
    optinal int age;
    required double score;
    repeated group hobbies{
        required string hobby_name;
        repeated string home_page;
    }
}

这里将兴趣列复杂了一些以展现parquet对嵌套的支持：

• Student做为整个schema的顶点，也是结构树的根节点，由message关键字标识；
• name做为必须有一个值的列，用required标识，类型为string；
• age做为可选项，能够有一个值也能够没有，用optinal标识，类型为string；
• score做为必须有一个值的列，用required标识，类型为double；
• hobbies做为能够没有也能够有多个的列，用repeated标识，类型为group，也就是嵌套类型；
  • hobby_name属于hobbies中元素的属性，必须有一个，类型为string；
  • home_page属于hobbies中元素的属性，能够有一个也能够没有，类型为string；

能够看到Parquet的schema结构中没有对于List、Map等类型的支持，事实上List经过repeated支持，而Map则是经过group类型支持，举例说明：

经过repeated支持List：

[15,16,18,14]
==>
repeated int ages;

经过repeated+group支持List[Map]：

{'name':'李四','age':15}
==>
repeated group Peoples{
    required string name;
    optinal int age;
}

存储格式

从schema树结构到列存储；

仍是上述例子，看下schema的树形结构：

矩形表示是一个叶子节点，叶子节点都是基本类型，Group不是叶子，叶子节点中颜色最浅的是optinal，中间的是required，最深的是repeated；

首先上述结构对应的列式存储总共有5列（等于叶子节点的数量）：

Column	Type
Name	string
Age	int
Score	double
hobbies.hobby_name	string
hobbies.page_home	string

Definition level & Repeatition level

解决上述歧义问题是经过定义等级和重复等级来完成的，下面依次介绍这两个比较难以直观理解的概念；

Definition level 定义等级

Definition level指的是截至当前位置为止，从根节点一路到此的路径上有多少可选的节点被定义了，由于是可选的，所以required类型不统计在内；

若是一个节点被定义了，那么说明到达它的路径上的全部节点都是被定义的，若是一个节点的定义等级等于这个节点处的最大定义等级，那么说明它是有数据的，不然它的定义等级应该更小才对；

一个简单例子讲解定义等级：

message ExampleDefinitionLevel{
    optinal group a{
        required group b{
            optinal string c;
        }
    }
}

Value	Definition level	说明
a:null	0	a往上只有根节点，所以它最大定义等级为1，可是它为null，因此它的定义等级为0；
a:{b:null}	不可能	b是required的，所以它不可能为null；
a:{b:{c:null}}	1	c处最大定义等级为2，由于b是required的不参与统计，可是c为null，因此它的定义等级为1；
a:{b:{c:"foo"}}	2	c有数据，所以它的定义等级就等于它的最大定义等级，即2；

到此，定义等级的计算公式以下：当前树深度 - 路径上类型为required的个数 - 1（若是自身为null）；

Repetition level 重复等级

针对repeated类型field，若是一个field重复了，那么它的重复等级等于根节点到达它的路径上的repeated节点的个数；

注意：这个重复指的是同一个父节点下的同一类field出现多个，若是是不一样父节点，那也是不算重复的；

一样以简单例子进行分析：

message ExampleRepetitionLevel{
    repeated group a{
        required group b{
            repeated group c{
                required string d;
                repeated string e;
            }
        }
    }
}

Value	Repetition level	说明
a:null	0	根本没有重复这回事。。。。
a:a1	0	对于a1，虽然不是null，可是field目前只有一个a1，也没有重复；
a:a1 a:a2	1	对于a2，前面有个a1此时节点a重复出现了，它的重复等级为1，由于它上面也没有其余repeated节点了；
a1:{b:null}	0	对于b，a1看不到a2，所以没有重复；
a1:{b:null} a2:{b:null}	1	对于a2的b，a2在a1后面，因此算出现重复，b自身不重复且为null；
a1:{b:{c:c1}} a2:{b:{c:c2}}	1	对于c2，虽然看着好像以前有个c1，可是因为他们分属不一样的父节点，所以c没有重复，可是对于a2与a1依然是重复的，因此重复等级为1；
a1:{b:{c:c1}} a1:{b:{c:c2}}	2	对于c2，他们都是从a1到b，父节点都是b，那么此时field c重复了，c路径上还有一个a为repeated，所以重复等级为2；

这里可能仍是比较难以理解，下面经过以前的张三李四的例子，来更加真切的感觉下在这个例子上的定义等级和重复等级；

张三李四的定义、重复等级

Schema以及数据内容以下：

message Student{
    required string name;
    optinal int age;
    required double score;
    repeated group hobbies{
        required string hobby_name;
        repeated string home_page;
    }
}

Student 1:
    Name 张三
    Age 15
    Score 70
    hobbies
    	hobby_name 篮球
        page_home nba.com
    hobbies
    	hobby_name 足球
    
Student 2:
    Name 李四
    Score 75

name列最好理解，首先它是required的，因此既不符合定义等级，也不符合重复等级的要求，又是第一层的节点，所以所有都是0；

name	定义等级	重复等级
张三	0	0
李四	0	0

score列所处层级、类型与name列一致，也所有都是0，这里就不列出来了；

age列一样处于第一层，可是它是optinal的，所以知足定义等级的要求，只有张三有age，定义等级为1，路径上只有它本身知足，重复等级为0；

age	定义等级	重复等级
15	1	0

hobby_name列处于hobbies group中，类型是required，篮球、足球定义等级都是1（自身为required不归入统计），父节点hobbies为repeated，归入统计，篮球重复等级为0，此时张三的数据中还没有出现过hobby_name或者hobbies，而足球的父节点hobbies重复了，而hobbies路径上重复节点数为1，所以它的重复等级为1；

hobbies.hobby_name	定义等级	重复等级
篮球	1	0
足球	1	1

home_page列只在张三的第一个hobbies中有，首先重复等级为0，这点与篮球是一个缘由，而定义等级为2，由于它是repeated，路径上它的父节点也是repeated的；

hobbies.home_page	定义等级	重复等级
nba.com	2	0

到此对两个虽然简单，可是也包含了Parquet的三种类型、嵌套group等结构的例子进行了列式存储分析，对此有个基本概念就行，其实就是两个等级的定义问题；

文件格式

Parquet的文件格式主要由header、footer、Row group、Column、Page组成，这种形式也是为了支持在hadoop等分布式大数据框架下的数据存储，所以这部分看起来总让人想起hadoop的分区。。。。。。

结合下面的官方格式展现图：

能够看到图中分为左右两部分：

• 左边：
  • 最外层表示一个Parquet文件；
  • 首先是Magic Number，用于校验Parquet文件，而且也能够用于表示文件开始和结束位；
  • 一个File对应多个Row group；
  • 一个Row group对应多个Column；
  • 一个Column对应多个Page；
  • Page是最小逻辑存储单元，其中包含头信息、重复等级和定义等级以及对应的数据值；
• 右边：
  • Footer中包含重要的元数据；
  • 文件元数据包含版本、架构、额外的k/v对等；
  • Row group元数据包括其下属各个Column的元数据；
  • Column的元数据包含数据类型、路径、编码、偏移量、压缩/未压缩大小、额外的k/v对等；

文件格式的设定一方面是针对Hadoop等分布式结构的适应，另外一方面也是对其嵌套支持、高效压缩等特性的支持，因此以为从这方面理解会更容易一些，好比：

• 嵌套支持：从上一章节知道列式存储支持嵌套中Repetition level和Definition level是很重要的，这两者都存放于Row group的元数据中；
• 高效压缩：注意到每一个Column都有一个type元数据，那么压缩算法能够经过这个属性来进行对应压缩，另外元数据中的额外k/v对能够用于存放对应列的统计信息；

Python导入导出Parquet格式文件

最后给出Python使用Pandas和pyspark两种方式对Parquet文件的操做Demo吧，实际使用上因为相关库的封装，对于调用者来讲除了导入导出的API略有不一样，其余操做是彻底一致的；

Pandas:

import pandas as pd
pd.read_parquet('parquet_file_path', engine='pyarrow')

上述代码须要注意的是要单独安装pyarrow库，不然会报错，pandas是基于pyarrow对parquet进行支持的；

PS：这里没有安装pyarrow，也没有指定engine的话，报错信息中说能够安装pyarrow或者fastparquet，可是我这里试过fastparquet加载个人parquet文件会失败，个人parquet是spark上直接导出的，不知道是否是两个库对parquet支持上有差别仍是由于啥，pyarrow就能够。。。。

pyspark:

from pyspark import SparkContext
from pyspark.sql.session import SparkSession

ss = SparkSession(sc)
ss.read.parquet('parquet_file_path') # 默认读取的是hdfs的file

pyspark就直接读取就好，毕竟都是一家人。。。。

参考

文中的不少概念、例子等都来自于下面两篇分享，须要的同窗能够移步那边；

• https://www.jianshu.com/p/47b39ae336d5
• https://zhuanlan.zhihu.com/p/141908285