SPARK Parquet嵌套类型的向量化支持以及列索引(column index)

Posted 2022-12-07 鸿乃江边鸟

背景

本文基于Spark 3.3.0
列式存储Parquet文件越来越受到工业界的青睐，在delta以及Spark中应用广泛，具体的项目见:parquet-mr

分析

Parquet格式

关于parquet的格式存储以及读取，可以参考大数据列存标准格式 - Parquet,总结一下就是：

Parquet采用类似Protobuf的协议来描述数据的Schema,字段的描述有三种（逻辑上）：

required  有且仅有一次
optional 0或1次
repeated 0次或多次

具体到物理存储，就得有Repetition Level（对应repeated）,Definition Level(对应optional) ,required是不需要的，因为字段存在就有，不存在就没有。
其实这种很好理解，因为在Dremel/Parquet中，提出的是以树状形式来组织schema中的字段，举例子：

message AddressBook 
  required string owner;
  repeated string phoneNumber;
  repeated group contacts 
    required string name;
    optional string phoneNumber;
  

对应到树形结构为：

              AddressBook
          /           |               \\  
         V            V                V
       owner      phoneNumber    contacts
                                    /    \\
                                   V      V
                                 name    phoneNumber 
                                

这样用Repetition Level 和Definition Level来表示数据位于哪一层级就能精确定位一个数据。

Parquet嵌套类型向量化

根据以上的Parquet的格式存储，在读取的Parquet文件的时候，对于非向量化的读取，是一行一行的读取，支持所有类型，对于向量化的读取在Spark 3.3.0以前是不支持嵌套类型（如 struct map list）的。
具体行读取如下（具体到page级别）：

向量化的读取如下：

可以看到是按照批次读取的。
注意：按照schema的定义，有些数据有可能是不存在的，所以对于同一列对应的Repetition Level 和Definition Level也有可能是不一样的。
性能也是有很大的提升：

Parquet的列索引

在Parquet 1.11.0之前，Parquet是不支持列索引的，具体见PARQUET-1201。
在Spark 3.2.0 便支持了paruqet的列索引的读取，具体见:SPARK-26345
在spark 3.2.0 之前Parquet的谓词下推是基于Row group的统计信息来的，如：最大最小值，字典信息，以及Parquet-1.12的Bloom filter，
在Spark 3.2.0 之后，我们可以基于page级别的数据过滤（只选择需要的page），这样能大大减少IO，因为在page级别过滤的话，不需要每次都会获取整个Row group的数据。

具体的实现逻辑如下:

之前的读取是把对应Row group的数据全部读取过来，之后再进行过滤。