哔哩哔哩基于 Iceberg 的智能数据组织优化实践

智能优化背景

首先来介绍一下智能优化的背景。

1. 湖仓一体架构与现状

我们的湖仓一体平台使用 Iceberg 作为数据的存储格式，数据存储于 HDFS，入湖主要有3 条链路：离线场景使用 Spark 写入数据，实时场景则使用 Flink 或我们提供的 Java SDK 进行写入。交互式分析采用 Trino 查询引擎，并利用 Alluxio 对数据进行缓存加速。此外，平台也有一独立的服务 Magnus 负责 Iceberg 表的数据优化，将在本次介绍中重点提及。对于写入 Iceberg 的数据，一部分会继续写入下游的 Iceberg 表，而在某些对查询性能和稳定性要求较高的场景，需要毫秒级响应时间，这时数据会被导出到 ClickHouse 或 ES。目前，平台包含大约 2000 张 Iceberg 表，总数据量达到 40PB，每日增量约为 100TB。Trino 的日查询量超过 400 万次，P99 的响应时间大约为 3 秒。

2. OLAP 场景查询加速

目前的业务场景主要包括 BI 报表、指标服务、A/B Test 人群筛选以及日志处理。针对这些场景，我们在 Iceberg 基础上进行了功能拓展，以满足用户对查询加速的需求。除了 Iceberg 原有的数据组织分布能力外，还增加了支持多维排序的功能，比如 Z-order 和 Hilbert Curve，并且提供了多种索引类型，包括通用的 Bloom Filter、Bitmap 以及针对日志场景的特殊索引。此外，我们还开发了预计算功能，主要用于加速聚合计算的查询。

3. 用户使用门槛高

我们支持的这些查询加速手段能够为查询带来数倍到数十倍的性能提升。然而，实际落地时会遇到一个问题，即这些优化手段对用户的使用门槛较高，要求用户对业务的查询模式有清晰的认知，并了解相关的基础知识才能进行合理配置。因此，通常需要我们手动为用户配置，或者通过组织培训提供指导。这限制了查询优化技术的推广使用。我们希望通过自动化、智能化的方式解决用户使用门槛的问题。这也是智能优化的目标，我们希望它能够自动分析用户的历史查询，为 Iceberg 表配置合理的优化手段，从而实现后续用户查询的加速。

02

接下来将介绍智能优化的整体实践方案。

1. 整体方案设计

整个智能优化流程涉及两个计算引擎，一个是交互式分析的 Trino 引擎，另一个是执行数据优化的 Spark 引擎。还包括两个服务，一个是查询采集服务，另一个是负责分析推荐和数据优化调度的 Magnus 服务。用户提交查询后，查询采集服务实时从 Trino 获取查询信息，并通过实时链路写入查询明细的 Iceberg 表。Magnus 的分析推荐模块定期从查询明细表获取查询信息，分析查询模式，生成推荐的数据组织配置，然后应用到 Iceberg 表中。这时推荐的配置并未实际生效，需要数据优化模块在 Iceberg 表数据写入后异步提交优化任务，由 Spark 完成数据优化。接下来会详细介绍查询采集、分析推荐以及数据优化三个模块的一些细节。

2. 查询信息采集

查询采集模块需要采集四类查询信息。首先是基本信息，包括时间、状态、查询 SQL、以及用户身份等。接着是反映查询性能的关键指标，如查询耗时和扫描数据量，这些指标直接影响交互式分析的用户体验和查询优化效果。第三个部分是查询模式，包括查询的过滤条件、Order by 条件以及聚合模型。这些信息理论上可以通过直接分析查询 SQL 得到，但我们选择将提取查询模式的工作放到 Trino 中实现，通过查询信息方式暴露出来，从而降低开发成本。最后是数据过滤的指标，我们在 Trino 统计了排序等优化手段在查询中的过滤效果，这些指标可以帮助我们跟踪和分析 Iceberg 表的优化效果，对推荐策略进行调整。

3. 分析推荐

分析推荐模块是根据查询以及 Iceberg 表的统计信息进行推荐的。我们的推荐策略是由一系列基于优化原理和实践经验的规则构成的。因此，在实现上相对比较简单，并且对于不是特别复杂的查询模式，也会有较好的推荐效果。分析推荐任务是定期执行的，比如每周执行一次针对每张 Iceberg 表的分析推荐任务。当用户的查询场景发生变化时，推荐配置也可以相应调整，以达到最佳优化效果。不同优化手段的分析对象和推荐依据是有区别的，下面将简要介绍各个优化手段的分析和推荐逻辑。

进行数据分析时，一个重要的优化策略是通过调整 Iceberg 表的分布来提高数据在特定字段上的聚集性。在查询过程中，引擎可以利用 Iceberg 表文件级别的最大值和最小值统计信息来过滤文件，实现查询加速。举个例子，假设一张表有四个文件，第一个文件在字段 a 上的最小值是 0，最大值是 10。如果查询条件是 a 等于 11，由于 11 不在 0 到 10 的范围内，该文件就无需被读取。相反，如果查询条件是 a 等于 2，而没有进行数据分布优化，就需要读取所有文件，因为每个文件在字段 a 上的最小值和最大值范围都包括 2。

对表按照字段 a 进行线性分布优化后，可以看到整个表在字段 a 上的聚集性得到改善。这样一来，只需读取第二个文件即可完成查询，其他文件则被直接过滤掉。在进行数据分布优化时，主要考虑常用的过滤字段，根据查询条件统计出每个字段出现在过滤条件中的查询占比，推荐优化策略更倾向于选择占比较高的字段。同时，也会考虑字段基数和分区文件数等因素。字段基数越高，分布效果就会更好。举个例子，如果将字段 a 的基数改为 2，即只有 0 和 2 两个取值各占一半，即使按照字段 a 进行分布优化，也至少需要读取两个文件。同理，如果分区文件数过少，分布优化效果也会变差。

在分布优化推荐中，有几种常见情况可以作为例子。首先，如果有 90% 的查询是针对 a 字段进行过滤，而只有 10% 的查询是针对 b 字段进行过滤，这种情况下推荐按照 a 字段进行线性分布。其次，如果有 50% 的查询是针对 a 字段进行过滤，而另外 50% 的查询同时针对 a 和 b 字段进行过滤，那么建议按照 a 和 b 字段的顺序进行线性分布，可以较好地过滤两类查询。第三种情况是 50% 的查询对 a 字段过滤，另外 50% 的查询对 b 字段过滤。在这种情况下，如果按照 a 或 b 字段线性分布，则对另一类查询都不会有很好的过滤效果。这时候可以考虑采用 Hilbert Curve 这种多维分布方式，它能够在多个字段上实现较好的聚集效果，提升不同字段过滤查询的性能表现。

索引是用于实现查询加速的一种重要技术，它和分布优化一样通过文件级别的数据过滤加速查询。索引提供了更为详尽的记录，因此其过滤性能更佳，能够处理一些分布优化效果不好的场景。例如当查询条件改为“a in (3,6,9)”时，尽管做了分布优化，仍需读取三个文件，实际上只需读取其中一个文件，这时就需使用索引来实现优化。此外，当参与分布的字段过多时，分布的效果可能较差，因此选择少量字段用于分布，而对其他字段进行索引构建，也是常见的优化策略。

不同类型的索引适用于不同的场景，需要综合考虑字段的过滤查询占比和不同类型的过滤条件，如等值过滤和范围过滤。Bloom Filter 适用于等值过滤，而针对范围过滤的查询，需使用 bitmap 索引。此外，字段基数也是重要的推荐依据，对于基数较高的字段，使用 Bloom Filter 索引效果较好，而构建 bitmap 索引可能会因为索引过大，导致性能回退。

文件内排序也是查询优化的常见手段，可以加速 Top N 查询。Top N 查询在计算引擎中执行时一般会分为局部排序和全局排序两个阶段。通过事先对 Iceberg 表文件内部进行排序，就可以节省局部排序的计算成本，同时减少扫描的数据量。

排序的推荐策略中，我们关注 Top N 查询中 Order by 子句的条件，根据 Order by 条件在查询中的占比来生成推荐。此外，需要注意同一种排序定义是可以响应多种 Order by 条件的查询的。举个例子，如果按照字段 a、b 进行排序，既可以响应按 a、b 排序的查询，也可以响应按照 a 排序的查询。这个因素在推荐的时候也是需要考虑的。

我们实现的预计算是一种针对聚合计算的优化手段。该手段通过提前按照指定的聚合模型对每个数据文件生成预聚合文件，在查询时直接读取这些预聚合文件，从而减少需要读取的数据量。预计算的分析过程相对复杂，首先需要定义聚合模型，然后计算每个模型在查询中的占比。聚合模型包括维度字段和聚合函数，对于涉及多表关联的场景，还需要考虑关联信息。另外，预计算的聚合效果也是重要的考量因素。如果一个文件经过聚合后仍然保留了大部分数据，那么预计算的意义就不大，同时还会浪费存储空间。这个指标通常无法从表的统计信息中获得，而需要通过额外的计算方法，比如通过 Trino 的查询来获取聚合效果。

推荐完成后，服务会将推荐结果配置到 Iceberg 表中。我们还做了推荐结果的持久化和前端展示，这样方便我们和用户查看。在生成的生产表的推荐记录中，展示了推荐的排序和分布信息。这个展示内容还包括推荐结果和当前配置的对比。例如，在 Distribution 这个部分，黑色字段表示保持不变，绿色字段表示新增，红色字段表示移除。通过这样的展示和对比，用户可以清晰地了解推荐结果和当前配置的变化。

4. 数据优化

推荐结果配置到 Iceberg 表之后，并不会立即生效，需要通过数据优化模块，异步提交 Spark 任务完成数据优化。数据优化的流程如上图，当任务向 Iceberg 表写入数据时，会发送一个 Commit Event。通过 Commit Event，调度器可以获取这次写入操作修改了哪些分区，写入了多少文件以及每个文件的数据量等信息。基于这些信息，以及 Iceberg 表的元数据信息，调度器会调度优化任务到 Spark 完成优化。

对于异步的数据优化，延迟是重要的考量指标。以小时分区的实时表为例，如果在分区所有数据写入完成再进行优化，可能导致超过一小时的延迟。用户查询过去 15 分钟至半小时内的数据时，性能就会比较差。为降低实时优化延迟，我们采用了基于快照的分级优化调度策略。Iceberg 表的快照是表在某个时间点的副本，每次提交时会生成一个快照，其中包含已存在的数据文件、本次提交的增量数据文件以及删除的数据文件。我们的策略是只优化指定快照中新增的数据，当某分区的小文件数量累积较多时，将触发 minor 优化以合并小文件。当累积未优化文件数据量达到阈值（如 2GB），将触发 major 优化，包括排序、分布以及索引创建等操作。这种策略将整个分区的优化拆分成多阶段优化，用户在查询时能享受到阶段性优化带来的加速效果。

为了降低优化延迟，我们还做了其他优化。一是任务合并，将同一张表的多个优化任务打包提交，减少 Spark 任务调度开销；二是通过优先级调度防止历史数据回刷对实时数据优化造成影响。此外，我们还针对优化任务进行了一些资源管理控制，如限制总体计算资源和单表的并发控制等。目前在高并发提交场景下，Iceberg 表存在性能问题，因此需要对每个表同时运行的任务数量进行限制。

我们还有一个前端页面，用来展示 Iceberg 表的分区级别的统计信息。除了数据量、文件数等基本信息，页面还展示最后写入时间以及各种优化手段的实际优化比例，比如排序和分布等，以便进行问题排查。

03

最后介绍一下智能优化现阶段的成果以及未来的规划。

1. 成果

我们目前针对一些没有进行任何优化配置的 Iceberg 表，开放了智能优化功能。截至目前为止，已经对 30 多张表进行了优化。在这 30 多张表经过优化后的 30 天总体扫描数据量减少了 28%。其中有超过 60% 的表扫描量减少了 30% 以上。目前项目的推荐策略还相对比较保守。在实际的生产环境中，有许多表已经由用户配置了一些优化手段，但由于配置不够合理，所以无法达到良好的加速效果。对这部分表开启智能优化功能，查询加速的收益会更高。

2. 未来规划

在接下来的工作中，我们将持续改进并推广智能优化功能。其中一项改进是增加推荐准确性。比如分布和索引的推荐，影响推荐准确率的关键因素之一是过滤效果的判断，因此我们会考虑使用更详细的统计信息，如实际数据分布，来辅助推荐决策。随着参考统计信息的增加，决策模型的参数将变得更加复杂，我们也将考虑利用机器学习或人工智能算法来进一步提高推荐准确性。另一个改进方向是支持更多的查询场景。目前索引和预计算推荐在日志场景和多表关联预计算推荐方面仍有不足，我们将逐步完善这些场景。最终，我们还将把智能优化推广应用到更多的生产表中，以优化用户配置并提供更好的查询体验。