查询毫秒时延的Parquet

通过拉斐尔Taylor-Davies安德鲁·兰姆/ 2022年12月7日/InfluxDB IOx，社区

我们相信查询数据Apache拼花直接文件可以实现与大多数专用文件格式类似或更好的存储效率和查询性能。虽然它需要大量的工程工作，但Parquet的开放格式和广泛的生态系统支持的好处使它成为广泛类型的数据系统的明显选择。

在本文中，我们将解释在Parquet格式中快速查询存储在Parquet格式中的数据所需的几种高级技术Apache Arrow Rust Parquet阅读器．这些技术使Rust实现成为查询Parquet文件的最快实现之一——无论是在本地磁盘上还是在远程对象存储上。它能够查询gb的拼花在一个以毫秒为单位．

背景

Apache拼花一种越来越流行的开放存储格式分析数据集，并且已经成为具有成本效益的、与dbms无关的数据存储的事实上的标准。Parquet最初是为Hadoop生态系统创建的，由于其引人注目的组合，现在它的范围广泛扩展到数据分析生态系统:

  • 高压缩比
  • 对商品blob-storage(例如S3)的适应性
  • 广泛的生态系统和工具支持
  • 跨许多不同平台和工具的可移植性
  • 支持任意结构的数据

越来越多的其他系统，如DuckDB和红移允许直接查询存储在Parquet中的数据，但与本地(自定义)文件格式相比，支持仍然是次要考虑因素。这些格式包括DuckDB.duckdb文件格式，Apache IOTTsFile,大猩猩的格式等。

以前只能在闭源商业实现中使用的相同复杂的查询技术现在第一次以开放源码的形式提供。所需的工程能力来自大型的、运行良好的、有全球贡献者社区的开源项目，例如Apache箭头和Apache黑斑羚．

Parquet文件格式

在深入研究有效读取的细节之前拼花，了解文件布局非常重要。文件格式经过精心设计，可以快速定位所需的信息，跳过不相关的部分，并有效地解码剩余的内容。

  • Parquet文件中的数据被分成称为rowgroup的水平片
  • 每个RowGroup为模式中的每一列都包含一个ColumnChunk

例如，下图演示了一个Parquet文件，其中“a”、“B”和“C”三列存储在两个rowgroup中，总共有6个columnchunk。

拼花文件格式图12.05.2022v1

ColumnChunk的逻辑值是使用其中之一写入的可用的编码依次添加到文件中的一个或多个数据页中。Parquet文件的末尾是一个页脚，它包含重要的元数据，例如:

  • 文件的模式信息，如列名和类型
  • RowGroup和columnchunk在文件中的位置。页脚还可以包含其他专门的数据结构:
  • 每个ColumnChunk的可选统计信息，包括最小/最大值和空计数
  • 指向的可选指针OffsetIndexes包含每个单独页面的位置
  • 指向的可选指针ColumnIndex包含每个Page的行计数和汇总统计信息
  • 指向的可选指针BloomFilterData，它可以快速检查一个值是否存在于ColumnChunk中

例如，前面图中的2个Row Groups和6个columnchunk的逻辑结构可能存储在一个Parquet文件中，如下图所示(不是按比例)。ColumnChunks的页面放在前面，然后是页脚。数据、编码方案的有效性以及Parquet编码器的设置决定了每个ColumnChunk所需页面的数量和大小。在本例中，ColumnChunk 1需要2个页面，而ColumnChunk 6只需要1个页面。除了其他信息外，页脚还包含每个数据页的位置和列的类型。

Parquet文件格式图2 12.05.2022v1

在创建Parquet文件时，有许多重要的标准需要考虑，例如如何优化数据排序/集群，并将其结构为rowgroup和data page。这样的“物理设计”考虑因素是复杂的，值得用自己的系列文章来讨论，而不是在这篇博文中讨论。相反，我们关注的是如何使用可用的结构使查询非常快。

优化查询

在任何查询处理系统中，以下技术通常可以提高性能:

  1. 减少必须从二级存储传输来处理的数据(减少I/O)
  2. 减少解码数据的计算量(减少CPU)
  3. 交错/管道数据的读取和解码(提高并行性)

同样的原则也适用于查询Parquet文件，如下所述:

解码优化

Parquet实现了令人印象深刻的压缩比使用复杂的编码技术例如运行长度压缩、字典编码、增量编码等。因此，cpu绑定的解码任务可以支配查询延迟。Parquet阅读器可以使用许多技术来提高该任务的延迟和吞吐量，就像我们在Rust实现中所做的那样。

矢量解码

大多数分析系统一次解码多个值为柱状内存格式，如Apache Arrow，而不是逐行处理数据。这通常被称为矢量化或柱状处理，它是有益的，因为它:

  • 摊销分派开销以打开正在解码的列的类型
  • 通过从ColumnChunk中读取连续的值来改善缓存的局部性
  • 通常允许在一条指令中解码多个值。
  • 通过一个大的分配来避免许多小的堆分配，从而为可变长度类型(如字符串和字节数组)节省了大量资源

因此，Rust Parquet Reader实现了专门的解码器，用于将Parquet直接读入柱状内存格式(箭头数组)。

流媒体解码

哪些行存储在跨columnchunk的哪些Pages中之间没有关系。例如，第10000行的逻辑值可能在列A的第一页，在列B的第三页。

向量化解码最简单的方法，也是最初在Parquet解码器中实现的方法，是一次解码整个RowGroup(或ColumnChunk)。

然而，考虑到Parquet的高压缩比，单个RowGroup很可能包含数百万行。一次解码这么多行不是最优的，因为:

  • 需要大量的中间RAM:典型的为处理而优化的内存中格式，如Apache Arrow，需要的比parquet编码格式多得多。
  • 增加查询延迟:后续处理步骤(如过滤或聚合)只有在整个RowGroup(或ColumnChunk)解码后才能开始。

因此，最好的Parquet读取器通过按需生成可配置大小的批量行来支持“流式”数据输出。批处理大小必须足够大，以摊销解码开销，但又必须足够小，以有效地使用内存，并允许在后续批处理解码时并发地开始下游处理。

Parquet文件流解码图12.05.2022v1

虽然流不是一个复杂的特性，但解码的状态性质，特别是跨多列和任意嵌套的数据，其中行和值之间的关系不是固定的复杂中间缓冲需要大量的工程工作来正确处理。

字典维护

字典编码，也称为分类编码是一种不直接存储列中的每个值的技术，而是存储一个称为“字典”的单独列表中的索引。该技术实现了的许多好处第三范式对于具有重复值的列(low基数)，并特别有效的字符串列，如“City”。

ColumnChunk中的第一页可以是字典页，其中包含列类型的值列表。然后，这个ColumnChunk中的后续页面可以将索引编码到这个字典中，而不是直接编码值。

考虑到这种编码的有效性，如果Parquet解码器只是将字典数据解码为本机类型，那么它将低效地反复复制相同的值，这对于字符串数据来说尤其灾难性。为了有效地处理字典编码的数据，必须在解码期间保留编码。方便地，许多柱状格式，如箭头DictionaryArray，支持这样的兼容编码。

在读取箭头数组时，保留字典编码可以极大地提高性能，在某些情况下可以超过60 x，以及使用更少的内存。

保存字典的主要复杂因素是字典是按ColumnChunk存储的，因此字典在rowgroup之间会发生变化。读取器必须自动为跨多个RowGroup的批重新计算字典，同时还要优化批大小平均分配到每个RowGroup的行数的情况。此外，一个列可能只有部分字典编码，使实施更加复杂。关于这项技术及其并发症的更多信息可以在博客将此技术应用于c++ Parquet阅读器。

投影下推

最基本的Parquet优化，也是对Parquet文件最常描述的优化是投影下推，从而减少I/ o和CPU的需求。在这种情况下，投影意味着“选择部分但不是全部列”。考虑到Parquet组织数据的方式，只需读取和解码引用列所需的columnchunk就可以了。

例如，考虑一个窗体的SQL查询

SELECT B from table where A >

该查询只需要列A和B(而不是C)的数据，并且可以将投影“下推”到Parquet阅读器。

具体来说，使用页脚中的信息，Parquet读取器可以完全跳过为列C(本例中的ColumnChunk 3和ColumnChunk 6)存储数据的数据页的抓取(I/O)和解码(CPU)。

拼花文件投影下推图12.05.2022v1

谓词下推

类似于投影下推，谓词pushdown也避免了从Parquet文件中获取和解码数据，但是使用了过滤器表达式。这种技术通常需要与查询引擎更紧密地集成，例如DataFusion，以确定有效的谓词并在扫描期间计算它们。不幸的是，如果没有仔细的API设计，Parquet解码器和查询引擎可能会紧密耦合，从而阻止重用(例如，在中有不同的Impala和Spark实现)Cloudera Parquet谓词下推文档)．Rust Parquet阅读器使用RowSelectionAPI来避免这种耦合。

RowGroup修剪

许多基于Parquet的查询引擎支持最简单的谓词下推形式，它使用存储在页脚中的统计信息来跳过整个rowgroup。我们将此操作称为RowGroup修剪，它类似于分区修剪在许多经典的数据仓库系统中。

对于上面的示例查询，如果特定RowGroup中A的最大值小于35，则解码器可以跳过从中获取和解码任何columnchunk整个RowGroup。

Parquet文件RowGroup剪枝图12.05.2022v1

请注意，修剪最小值和最大值对许多数据布局和列类型都有效，但并非全部有效。具体来说，对于具有许多不同的伪随机值(例如标识符或uuid)的列来说，这种方法并不有效。值得庆幸的是，对于这个用例，Parquet还支持每个ColumnChunk布鲁姆过滤器．我们正在积极努力添加bloom滤镜支持Apache Rust的实现。

页面修剪

更复杂的谓词下推形式使用可选页面索引在页脚元数据中排除整个数据页。解码器只从其他列中解码相应的行，通常跳过整个页面。

由于各种原因，不同columnchunk中的页面通常包含不同数量的行，这使得这种优化更加复杂。虽然页面索引可以从一列中识别所需的页面，但从一列中删除页面并不会立即排除其他列中的整个页面。

页面修剪过程如下:

  • 结合使用谓词和页索引来标识要跳过的页面
  • 使用偏移索引来确定哪些行范围对应于未跳过的页面
  • 计算跨未跳过页面的范围的交集，并仅解码那些行

最后一点实现起来非常重要，特别是对于嵌套列表一行可以对应多个值．幸运的是，Rust Parquet阅读器在内部隐藏了这种复杂性，可以任意解码RowSelections．

例如扫描存储在5个数据页中的列A和列B，如下图所示:

如果谓词是A > 35，

  • 使用页面索引(最大值为20)修剪页面1，留下RowSelection[200->起]，
  • Parquet阅读器完全跳过第3页(因为它的最后一行索引是99)
  • (仅)通过读取第2、4和5页来读取相关行。

如果谓词改为A > 35 AND B = " F "，则页索引更有效

  • 使用> 35，生成RowSelection[200->起]
  • 使用B = " F "，在B剩下的第4页和第5页，产生一个RowSelection [100-244]
  • 相交两个RowSelection会得到一个合并的RowSelection [200-244]
  • Parquet阅读器只能从第2页和第4页解码这50行。

Telegraf配置

支持从Arrow c++和扩展pyarrow/pandas读取和写入这些索引拼花- 1404．

晚物质化

前两种形式的谓词下推只对解码值之前为rowgroup、columnchunk和Data page存储的元数据进行操作。但是，同样的技术也可以扩展到一个或多个列的值后解码他们，但在解码其他专栏之前，这通常被称为“后期物化”。

这种技巧在以下情况下尤其有效:

  • 谓词是非常有选择性的，即过滤掉大量的行
  • 每行都很大，要么是因为行很宽(例如JSON blobs)，要么是因为列很多
  • 所选数据被聚类在一起
  • 谓词所需的列解码成本相对较低，例如PrimitiveArray / DictionaryArray

有关于这种技术的好处的额外讨论火星- 36527和黑斑羚．

例如，给定谓词A > 35 AND B = " F "，其中引擎使用页面索引确定在RowSelection[100-244]中只有50行可以匹配，使用后期物化，Parquet解码器:

  • 解码列A的50个值
  • 在这50个值上计算A > 35
  • 在这种情况下，只有5行通过，导致RowSelection:
    • RowSelection (205 - 206)
    • RowSelection (238 - 240)
  • 对于这些选择，只解码列B的5行

拼花牌文件后期物化图12.05.2022v1

在某些情况下，例如B存储单个字符值的示例，后期物化机制的成本可能超过解码所节省的成本。然而，当上面列出的一些条件得到满足时，节省的费用是可观的。查询引擎必须决定下推哪些谓词，以及以何种顺序应用它们以获得最佳结果。

虽然这超出了本文的范围，但同样的技术可以应用于多个谓词以及多个列上的谓词。看到RowFilter界面在拼花板条箱获取更多信息，和row_filter在DataFusion中实现。

I / O叠加

而拼花的设计是为了高效的访问HDFS分布式文件系统，它非常适用于商品blob存储系统，如AWS S3，因为它们具有非常相似的特征:

  • 相对较慢的“随机访问”读取:在每个请求中读取大(mb)段的数据要比为较小的部分发出许多请求更有效
  • 检索第一个字节前的显著延迟
  • 每请求成本高:通常对每个请求收费，不管读取的字节数是多少，这鼓励每个请求读取大量连续的数据段。

为了从这样的系统中最佳地读取，Parquet阅读器必须:

  1. 尽量减少I/O请求的数量，同时应用各种下推技术以避免获取大量未使用的数据。
  2. 与适当的任务调度机制集成，在获取的数据上交织I/O和处理，以避免管道瓶颈。

由于这些都是重大的工程和集成挑战，许多Parquet阅读器仍然要求将文件完整地提取到本地存储。

获取整个文件来处理它们是不理想的，原因如下:

  1. 高延迟:在获取整个文件之前不能开始解码(Parquet元数据位于文件的末尾，因此解码器必须在解码其余部分之前看到结束部分)
  2. 浪费工作:获取整个文件将获取所有必要的数据，但也可能会获取大量不必要的数据，这些数据在读取页脚后将被跳过。这不必要地增加了成本。
  3. 需要昂贵的“本地连接”存储(或内存):许多云环境不提供带有本地连接存储的计算资源——它们要么依赖于昂贵的网络块存储，如AWS EBS，要么将本地存储限制为某些类别的虚拟机。

为了避免缓冲整个文件，需要一个复杂的Parquet解码器，它与I/O子系统集成在一起，可以首先获取和解码元数据，然后对相关数据块进行远程获取，与Parquet数据的解码交织在一起。这种优化需要仔细的工程，以从对象存储中获取足够大的数据块，使每个请求的开销不会从减少传输的字节中获益。火星- 36529更详细地描述顺序处理的挑战。

拼花文件IO下推图12.05.2022v1

此图中不包括合并请求和确保实际实现所需的最小请求大小等细节。

Rust Parquet板条箱提供了一个异步Parquet阅读器，以有效地读取任何AsyncFileReader:

  • 有效地从支持范围请求的任何存储介质读取
  • 与Rust的期货生态系统集成，避免阻塞等待网络I/O的线程可轻松实现CPU与网络的交叉
  • 同时请求多个范围，以允许实现合并相邻的范围，并行获取范围等。
  • 使用前面描述的下推技术尽可能地消除获取数据
  • 与Apache Arrow轻松集成object_store板条箱，你可以阅读更多在这里

为了说明可能发生的事情，下图显示了从远程文件获取页脚元数据的时间轴，使用该元数据确定要读取哪些数据页，然后同时获取数据和解码。为了匹配网络延迟、带宽和可用CPU，这个过程通常必须同时对多个文件执行。

Parquet文件IO下推图2 12.05.2022v1

结论

我们希望您喜欢阅读关于Parquet文件格式的内容，以及用于快速查询Parquet文件的各种技术。

我们相信，Parquet的大多数开源实现之所以没有本文中所描述的功能的广度，是因为它需要付出巨大的努力，而这在以前只有资金充足的商业企业才有可能实现它们的实现。

然而，随着Apache Arrow社区(Rust从业者和更广泛的Arrow社区)的增长和质量的提高，我们协作和构建前沿开源实现的能力是令人振奋和非常满意的。本博客中描述的技术是许多工程师贡献的结果，这些工程师分布在公司、爱好者和世界上的几个存储库中Apache Arrow DataFusion，Apache箭头和阿帕奇箭弩。

如果您有兴趣加入DataFusion社区，请保持联系．