使用 Parquet 的 Bloom Filter

作者：Trevor Hilton / 开发者
2024 年 5 月 28 日

注意：我们在 2024 年 6 月 6 日更新了这篇文章，以反映宝贵的社区反馈。

摘要

在本文中，我们探讨了何时以及如何在 Parquet 中使用 Bloom Filter，它们对写入的 Parquet 文件的影响，并衡量它们在处理大量高基数¹数据时的有效性。

总结来说，我们发现：

• 当处理大量高基数集群数据时，Bloom Filter 参数与每个 Parquet 行组中不同值的数量更紧密地匹配，可以产生最佳的剪枝效率。
• 使用此类 Bloom Filter，查询时间缩短至原来的 1/30，但每个列每个行组的存储空间成本为 2 KB 到 8 KB。
• 选择与整个数据集基数匹配的 Bloom Filter 参数会产生巨大的存储开销，但在我们的实验中，这并非必要。

包含不同输入参数计算出的 Bloom Filter 大小的表格可在Bloom Filter 大小中找到。

背景和动机

在查询 Apache Parquet 文件中包含的数据时，可以使用多种技术来加快查询速度。一种方法是使用 Bloom Filter 来剪枝整个行/列块。Bloom Filter 对于包含许多不同的伪随机值的列（例如，标识符或 UUID）特别有用。这与包含有序数据、基数较低或已通过行组元数据中的最小/最大统计信息有效剪枝的列不同。Apache Rust² 和 Java³ 实现的 Parquet 格式以及 Apache DataFusion 查询引擎均支持 Bloom Filter。

在 InfluxData，我们经常处理高基数数据，并正在考虑几种索引策略。鉴于 Bloom Filter 是 Apache Parquet 的一部分，以及我们用来支持它们的库，它们是一个有吸引力的选择。然而，缺乏关于它们在实践中的有效性或如何设置调整旋钮的信息。具体来说，我们想知道：

• Bloom Filter 会使 Parquet 文件的大小增加多少？
• 对于具有高基数的数据（例如，10 万到 100 万个不同的值），Parquet Bloom Filter 在多大程度上提高了查询性能？

Bloom Filter 的用例

考虑图 1 中的示例 Parquet 文件，该文件存储了在不同云区域托管的各种主机服务器上运行的容器的 CPU 使用率。

图 1：示例 Parquet 文件，其中包含按基数递增顺序排列的三列，以及时间和cpu列；region 的基数为 10，host 的基数为 1,000，container_id 的基数为 1,000,000。数据集的基数与基数最高的列的基数相对应，即 1,000,000。此 Parquet 文件由两个行组组成，每个行组都包含每列的最小值和最大值的元数据。

此 Parquet 文件中的行按 region、然后按 host、再按 container_id 排序，以匹配生成数据的实体的层次结构。实际上，对于 region、host 和 container_id 的每个唯一组合，都会有很多行，即表示不同时间戳的 CPU 使用率。

如果我们使用 region 和 container_id 作为谓词查询图 1 中的 Parquet 文件，例如：

SELECT * FROM 'fig1.parquet' WHERE region = 'us-east-2' AND container_id = 'CCC';

那么查询引擎将剪枝 Row Group 2，因为 us-east-2 超出了 Row Group 2 元数据中包含的 region 列的最小/最大值，因此只会解码 Row Group 1 中的行。但是，如果我们仅使用 container_id 进行查询，例如：

SELECT * FROM 'fig1.parquet' WHERE container_id = 'CCC';

那么查询引擎必须扫描两个行组，因为仅凭元数据无法根据其各自的 container_id 值的最小值/最大值排除任何一个组（值 CCC 落在 AAA..DDD 和 BBB..FFF 两个范围内）。

这就是 Bloom Filter 的用武之地。将 Bloom Filter 应用于 container_id 列，使查询引擎能够在将该列用作谓词时剪枝行组。

Parquet 的 Bloom Filter 实现

Parquet 使用 Split Block Bloom Filter (SBBF)。在写入 Parquet 数据时，通过指定以下参数，在每个列的基础上配置 SBBF：

• 不同值的数量 (NDV)
• 误报率 (FPP)

Bloom Filter 从 Parquet 文件中读取，使用其长度和偏移量，这些长度和偏移量存储在每个行组的元数据中，DataFusion 使用它们在查询文件时过滤掉整个行组。似乎合理的假设是，NDV 应该与它应用的列的基数紧密匹配，但是 NDV 越大，Bloom Filter 在文件中占用的空间就越多。

我们有兴趣了解低估 NDV 的影响，因为这可以节省存储空间。此外，我们假设选择的 FPP 应该与我们将从 Bloom Filter 中获得的剪枝量相对应。

Bloom Filter 大小

Bloom Filter 不是免费的。它们存储在 Parquet 文件中，因此会产生存储开销。所需 NDV 和 FPP 的计算决定了 Filter 的大小。FPP 可以计算为⁴

其中 f 是 FPP，k 是哈希函数的数量，n 是 NDV，m 是 Bloom Filter 中的总位数。要实现所需的 FPP 和 NDV，所需的位数可以通过重新排列上述公式来确定：

SBBF 使用八个哈希函数，以干净地适应 SIMD 通道⁵。因此，k 设置为 8。SBBF 将这些 m 位分散在一组块上，每个块的大小为 256 位，即 32 字节。Parquet Bloom Filter 的 Rust⁶ 和 Java⁷ 实现都将大小（以字节为单位）确定为：

m 除以 8 以字节表示。

使用上面的公式，我们可以计算各种 FPP 和 NDV⁸ 值的大小：

实验

设置

虽然上一节中的理论结果提供了信息，但我们希望验证它们是否与现实相符。为此，我们开发了 parquet-bloom-filter-analysis 工具。

它是一个 CLI，用 Rust 编写，用于生成具有指定列数的 Parquet 文件，每个列都具有：

• 基数，即唯一值的数量
• 可选的 Bloom Filter，带有 FPP 和 NDV 参数

此外，在生成一组文件时，我们指定：

• 要生成的 Parquet 文件数量
• 最大行组大小
• 重复因子

生成的行按基数递增的列排序，如上面的图 1 所示的示例。重复因子用于为生成的列值的每个唯一组合生成重复行，从而按列值的唯一组合对数据进行聚类。这类似于图 1 示例中每个 region/host/container 组合在不同时间戳记录的许多 CPU 使用率指标。

我们使用 parquet-bloom-filter-analysis 工具进行了以下实验：

1. NDV 对大小的影响：对于基数为 1M 和基数为 10 万的数据，生成少量具有固定 FPP 和不同 NDV 参数的文件，以确定每个基数下这些参数的可接受基线，并验证对 Parquet 文件大小的影响是否与理论相符。
2. 查询性能影响：为 1M 和 10 万基数生成更大容量的 Parquet 文件，以评估在更实际的规模下使用 Bloom Filter 可实现的查询性能提升。

系统规格

所有实验均在具有以下规格的机器上运行：

NDV 对大小的影响：1M 基数

我们生成了一个包含三列的数据集，具有以下属性：

这总共产生了 1 亿行数据，每个列值唯一组合有 100 行。

以下是生成的数据的示例：

+----------+-------------+--------------+
| col_0    | col_1       | col_2        |
+----------+-------------+--------------+
| first-52 | second-4119 | third-136666 |
| first-52 | second-4119 | third-230251 |
| first-52 | second-4119 | third-287794 |
| first-52 | second-4119 | third-360226 |
| first-52 | second-4119 | third-584614 |
| first-52 | second-4119 | third-672996 |
| first-52 | second-4119 | third-952788 |
| first-52 | second-4174 | third-002478 |
| first-52 | second-4174 | third-236114 |
| first-52 | second-4174 | third-439784 |
| first-52 | second-4174 | third-477628 |
| first-52 | second-4174 | third-540914 |
| ...                                   |
+----------+-------------+--------------+

从那里，我们通过将 FPP 设置为 0.01 并改变 NDV，在基数最高的列（即第 3 列）上创建 Bloom Filter。选择 0.01 的 FPP 是因为生成的数据仅包含 100 个行组，这应该对应于解码的行组减少两个数量级，即每 100 个行组中只有 1 个被解码将是最佳情况的剪枝效率。

存储成本可以通过取使用 Bloom Filter 时的平均 Parquet 文件大小与平均基线文件大小之差来衡量。此数据集的基线文件大小是通过在没有任何 Bloom Filter 的情况下编码相同数据来确定的。

为了查看实际剪枝了多少行组，我们使用了 Datafusion 的 CLI⁹ 工具来确定在使用对 Filtered 列的谓词进行查询时剪枝了多少行组，如下所示：

EXPLAIN ANALYZE SELECT * FROM '<data\_folder>/' WHERE col_2 = '<target_value>'; 

<data_folder> 对应于使用感兴趣的 FPP/NDV 生成的数据集的位置，并测试了代表性的 <target_value> 范围。

结果

基线存储成本，即没有 Bloom Filter 的平均 Parquet 文件大小为 647,080.5 B。以下是固定 FPP 为 0.01，不同 NDV 的结果：

差值 - 平均文件大小 与基线文件大小之间的差值；RG - 行组。

讨论/主要观察结果

• DataFusion 使用每个行组 8 KB 的 Bloom Filter / 文件大小增加 12% 来剪枝超过 90% 的行组。所有不匹配的行组都以每个组 16 KB 的开销 / 文件大小增加 20% 的代价被剪枝。
• 相对成本在这里看起来很显着，但在实践中会降低。例如，InfluxDB 生成的 Parquet 文件往往以 MB 为单位大小，并存储更多列，包括时间戳和 64 位浮点列。因此，我们预计每个行组增加 8-16 KB 将导致存储增长约 1% 或更少。
• 选择与列的实际基数紧密对应的 NDV 会产生巨大的存储成本（每个行组 2 MB），但不会提高剪枝效率，因此似乎没有必要。

NDV 对大小的影响：10 万基数

与之前的实验类似，‌此实验使用具有以下属性的数据：

有两个不同之处：重复因子为 1000（与 100 相比），并且第 3 列的基数为 10 万（与 1 百万相比）。这使行数和 Parquet 文件数与之前的实验保持一致，同时保持生成的数据有序。

文件大小和剪枝效率的测量方法与之前的实验相同。

结果

基线存储成本，即未添加 Bloom Filter 时的平均 Parquet 文件大小为 71,965.6 B。以下是固定 FPP 为 0.01，不同 NDV 的结果：

差值 - 平均文件大小 与基线文件大小之间的差值；RG - 行组。

讨论/主要观察结果

• 对文件大小的影响与之前的实验相同，与理论结果一致，即 Bloom Filter 大小与它正在过滤的数据的基数无关。
• DataFusion 在 NDV 为 1,000 时成功剪枝了所有不匹配的行组，每个行组仅增加了约 2K 的开销。
• 剪枝效率在比之前实验 (7,500) 更低的 NDV (1,000) 时达到饱和。

查询性能影响：10 万基数

为了评估 Bloom Filter 在更实际的规模下的性能，我们生成了具有以下属性的 Parquet 数据：

这对应于基数为 10 万的数据集，分布在 1 万个文件中。生成数据时，没有 Bloom Filter 以建立基线，并且在第 3 列上使用了 Bloom Filter，即使用FPP 为 0.01 和 NDV 为 1,000 的 10 万基数。

我们想在此场景中检查的两件事是：

1. 剪枝效率是否与上面执行的小规模实验保持一致？
2. 我们是否可以衡量在具有 Bloom Filter 的文件和没有 Bloom Filter 的文件之间查询性能的改进？

(1.) 可以通过分析查询以选择 ‌第 3 列的单个值来回答，而 (2.) 可以通过‌执行查询并测量响应时间来回答。

结果

该实验生成了 10,000 个文件，平均文件大小如下：

根据这些数字，每个行组的计算开销仍然是 2,076.8 B，与之前的实验相同；但是，基线是 9,112.87 B，而之前的基线是 71,965.60 B。

查询是使用第 3 列（即基数为 10 万的列）作为谓词执行的，值跨越其基数内的总值范围 (third-[00000, 99999])

RG - 行组。

讨论/主要观察结果

• Bloom Filter 的开销与较大的文件相同，但在规模上被放大。据推测，这是因为列式数据正在被压缩，而元数据中的 Bloom Filter 没有被压缩。
• Bloom 过滤器清理了行组统计信息未清理的行组，从而显著 улучшает 查询延迟（约 1/30）
• 仅基于行组统计信息，所有可用值的最小值/最大值的谓词值将提供快速查询。

查询性能影响：100 万基数

此实验使用了与先前示例相同的设置，生成了 10,000 个文件；但是，在本例中，第三列的基数增加到 100 万。调整了重复因子以保持生成的数据有序

FPP 和 NDV 参数也保持不变，分别为 FPP 0.01 和 NDV 1,000。 这是为了查看在此规模和基数下，是否可以使用较低的 NDV 以及因此较小的 Bloom 过滤器，尽管事实上在较小规模下，NDV 为 1,000 在 100 万基数下没有太大影响（请参阅先前的实验）。

结果

该实验生成了 10,000 个文件，平均文件大小如下

因此，Bloom 过滤器的开销为每个行组 2,076.8 B。

查询是使用第 3 列（即基数为 100 万的列）作为谓词执行的，其值跨越其基数内的总值范围（third-[000000, 999999]）

RG - 行组。

讨论/主要观察结果

• 尽管仅使用 1,000 的 NDV，Bloom 过滤器在此规模和基数下仍然有效

结论

当使用 Parquet 时，Bloom 过滤器可以在更大的数据量下提供显著的查询性能提升，尽管会增加额外的存储成本。好消息是，更适中的 Bloom 过滤器参数（在我们的有序数据中非常有效）可以减轻存储成本。

一些仍然悬而未决的问题是

• 是否可以使用更适中的 Bloom 过滤器参数，即 NDV < 1,000，以达到相同的效果，从而进一步减轻存储成本？
• 压缩写入的 Parquet 文件中的 Bloom 过滤器（目前不是 Parquet 规范的一部分）是否会带来任何好处？
• 如果我们对应用 Bloom 过滤器的列存储更长或更复杂的值（例如 UUID）时，是否能够获得相同的性能提升？

相关博客文章

• Apache Parquet
• Apache Parquet 简介
• 使用毫秒级延迟查询 Parquet
• Flight、DataFusion、Arrow 和 Parquet：使用 FDAP 架构构建 InfluxDB 3.0

1. 我们使用术语“基数”来指代整个数据集中的不同值的数量，即跨所有 Parquet 文件及其行组。为本文分析生成的数据是有序的，因此给定行组中不同值的数量小于整个数据集的基数。

2. 另请参阅 跟踪问题，了解 Apache Rust Parquet Bloom 过滤器支持

3. 另请参阅 博客文章，了解 Apache Spark 中的 Bloom 过滤器支持

4. 请参阅 共享内存机器的缓存高效 Bloom 过滤器

5. 请参阅 分割块 Bloom 过滤器

6. 请参阅 Rust Parquet 实现的最佳字节数 源代码

7. 请参阅 Java Parquet 实现的 源代码

8. 请参阅 Rust Playground 链接

9. 请参阅 DataFusion CLI