Uptycs: 构建快如闪电的分析

在 Uptycs，我们的数据平台架构多年来随着几乎所有数据平台的自然发展而发展。最初我们的架构围绕在线事务处理 （OLTP） 数据库 （在我们的例子中主要是 PostgreSQL）展开，用于管理以下类别的数据：

1. 1. 客户特定的配置
2. 2. 特征元数据
3. 3. 访问控制策略
4. 4. 其他微服务相关数据
5. 5. 微服务生成的数据

当数据访问简单时，这种设置效果很好，但随着我们平台的扩展，管理和利用这些数据（ 尤其是第 5 类数据）的复杂性也随之增加。工程师通常不得不手动查询单个数据库，编写自定义脚本来关联多个存储的数据，并处理缺乏数据全局视图的问题。

随着时间的推移，随着越来越多的团队构建自己的服务并引入针对其特定工作负载量身定制的新数据库，我们的数据变得越来越分散和碎片化。其中一些决策是有意为之的 — 由微服务最佳实践驱动，而另一些则是快速迭代和团队自主的自然结果。

与此同时我们正在构建 PB 级数据湖，作为存储大规模网络安全遥测和分析数据的基础。然而，随着数据生态系统的发展，我们面临着对跨数据库分析工作流日益增长的需求，这需要将 OLTP 数据与我们的数据湖连接起来 。这种碎片化造成了瓶颈，很明显，我们需要一种方法来跨多个来源无缝查询数据，而无需大量移动数据。

这促使我们探索了联合查询引擎 ，它允许跨不同数据源（例如 HDFS/S3、PostgreSQL、MongoDB）进行查询，而无需将所有内容迁移到单个系统中。

在 Trino、Apache Drill 和 Dremio 等可用选项中，我们最终选择了 Trino（以前称为 PrestoSQL） 作为我们的主要联合查询引擎，因为它具有卓越的性能、可扩展性、ANSI SQL 合规性和广泛的连接器生态系统。

Trino 的分布式 SQL 执行使我们能够高效地跨多个 OLTP 数据库和数据湖运行查询，而无需昂贵的 ETL 管道。

我们当前的数据平台架构与 Trino 联合查询引擎

展望未来，联合查询成为我们数据平台的基石，支持跨不同存储的无缝数据访问，同时保持高性能和可扩展性。但是随着 OLTP 数据库和数据湖中的数据量激增，我们遇到了性能瓶颈，尤其是在 OLTP 数据库和数据湖之间运行复杂的 JOIN 作时。跨多个数据存储实时查询的开销变得不可持续，这促使我们重新考虑我们的方法。

现有方法的挑战

随着数据平台的扩展，我们在处理大规模分析查询的同时确保事务工作负载保持性能方面面临着越来越多的挑战。核心问题源于 OLTP 数据库（如 PostgreSQL）和分析查询引擎（如 Trino）的运行方式的根本差异。

1. 1. PostgreSQL 中的读写争用 PostgreSQL 充当主要的事务存储，处理实时读取和写入密集型工作负载。随着 PostgreSQL 上的分析查询增加，OLTP 和 OLAP 工作负载之间的争用成为严重的瓶颈。 为了缓解这种情况，我们最初引入了 PostgreSQL 的只读副本，并将 Trino 查询重定向到它们。然而这种方法带来了新的挑战
   • • 复制滞后： 主数据库上的繁重写入工作负载会导致频繁的复制延迟，从而导致只读副本的查询结果过时或不一致。
   • • 长时间运行的查询的影响： Trino 的分析查询处理了数百万条记录，使复制滞后变得更糟。只读副本上长时间运行的查询经常与复制过程冲突，经常触发臭名昭著的 “cancelling statement due to conflict with recovery” 错误。增加复制延迟参数可能会暂时减少查询取消，但代价是复制滞后严重，从而延迟对新数据的访问。
   • • 不断增长的写入延迟： 随着传入数据量的激增，PostgreSQL 写入作变得更慢。管理索引、清理和维护数据库运行状况在作上变得越来越复杂。
2. 2. 跨系统查询中的性能瓶颈 一个主要的痛点是运行联合查询，将 PostgreSQL 中的结构化 OLTP 数据与 Hive 中基于 ORC/Parquet 的大量表联接起来。查询数十亿条记录的控制面板速度明显变慢，主要是由于 Trino 中 PostgreSQL 连接器的限制。
   • • 单节点执行瓶颈： 与利用分布式查询执行和基于文件的并行化拆分的 Hive 不同，PostgreSQL 的 Trino 连接器通过 JDBC 按顺序处理数据。这导致了高网络开销、有限的查询并发性（由于 PostgreSQL 的连接限制）并增加了 Trino 的内存压力
   • • 缺乏下推优化： 虽然 Trino 可以将过滤器和投影下推到 PostgreSQL，但它无法在 PostgreSQL 和 Hive 之间下推 JOIN。这迫使将大型数据集提取到 Trino 中进行内存处理，从而导致执行时间和资源消耗增加
   • • 缺少并行度： 与 Trino 的 Hive 连接器不同，后者可以有效地将大型数据集拆分到多个工作节点， 从而支持从 ORC/Parquet 文件并行读取，而 Postgres 连接器使用 JDBC 连接按顺序检索数据，受数据库的单节点执行模型的约束。
   • • 多源查询中的延迟： 许多控制面板查询多个 PostgreSQL 数据库以及 HDFS/S3 中的数据。跨这些系统编排查询，每个系统都有截然不同的延迟和吞吐量，这进一步降低了性能。
3. 3. 维护单一事实来源的挑战 考虑到这些挑战，一种可能的方法是将所有微服务直接指向数据湖 ，从而消除对联合查询的需求。

然而这远非一个简单的解决方案。这需要大量的迁移工作，重写无数的服务和查询，适应新的查询范式，并确保与现有工作流的兼容性，所有这些都在保持运营稳定性的同时。

另一种选择是通过 ETL 管道定期将分析数据从 OLTP 数据库迁移到数据湖 。虽然这种方法适用于某些批处理工作负载，但它引入了数据新鲜度问题、PostgreSQL 中的更新与其在数据湖中的可用性之间的延迟。

我们的许多微服务都依赖于通过 REST API 提供的实时事务数据，而引入定期同步意味着洞察可能会过时几分钟甚至几小时，从而导致运营工作负载和分析工作负载之间不一致。

随着我们对实时洞察的需求不断增长，这两种解决方案都被证明是不够的，这促使我们探索一种更加无缝和高效的替代方案。

需要可扩展的解决方案

鉴于这些挑战，我们需要一种混合解决方案 ，该解决方案能够：

• • 将经常查询的数据从 PostgreSQL 卸载到可扩展的存储层，以减少数据库争用。
• • 确保事务和分析系统之间的数据一致性 ，避免复制滞后和过时的读取。
• • 利用数据湖进行大规模分析查询 ，而不会中断实时运营工作流程

这就是我们转向统一、可扩展的数据管道的原因，利用 Debezium 和 Apache Hudi 的变更数据捕获 （CDC）。

为什么选择 Debezium，什么是变更数据捕获 （CDC）？

对于初学者来说，CDC 是一种通过复制槽跟踪源数据库中的更改（插入、更新、删除）并将其实时流式传输到下游系统的技术。它支持高效的数据复制，无需全表扫描或定期批处理作业。

为什么选择 Debezium？

Debezium 是一个开源 CDC 工具，可与 PostgreSQL、Kafka 和其他事件流平台无缝集成。主要优势包括：

• • 基于日志的 CDC，性能开销最小
• • Schema 演变支持
• • 容错和至少一次交付（也可以使用 Kafka Connect 实现恰好一次交付）
• • 适用于 PostgreSQL 数据库的预构建连接器，能够大规模流式传输更改事件

CDC Ingestion Pipeline 架构概述

我们基于 CDC 的摄取管道的高级架构如下：

第 1 部分：

1. 1. PostgreSQL — 事务数据所在的源数据库。（多个数据库实例）
2. 2. Debezium — 通过复制槽从 PostgreSQL WAL 捕获更改，并将其流式传输到 Kafka。

第 2 部分：

1. 1. Kafka — 充当事件流和容错的缓冲区。
2. 2. DebeziumConsumer — 装饰器服务负责从多个特定于表的 Kafka 主题中读取数据，根据我们的通用 Hudi 架构对其进行装饰，并进一步丰富通过 Debezium 传入的事件的数据，最终将装饰的事件发布到单个 Kafka 主题，Spark 作业将通过该主题选择此类事件进行摄取。它处理其他几个使用案例，例如其他微服务使用的异步缓存同步。

第 3 部分：

1. 1. Spark Streaming App — 通过 Apache Hudi 处理数据湖中的更新插入和增量更新
2. 2. 数据湖 （S3/HDFS） — 将原始数据和转换后的数据存储在 parquet/ORC 文件中
3. 3. 查询层 — Trino，支持对提取的数据进行分析和查询

通过实施 CDC 到 Hudi 摄取管道 ，我们已将分析数据整合到一个优化的存储层中，从而能够从 HDFS 或 S3 或任何其他对象存储高效处理所有读取工作负载。利用结构化分区和 Bloom 筛选器 ，我们在 Trino 上的控制面板查询性能提高了 **10 倍 **。

通过以最小（可观察和可作的）CDC 延迟将数据卸载到基于 Hudi 的数据湖 ，我们克服了在不同数据库上运行分析工作负载的所有挑战。Trino 中的 Hive 和 Hudi 连接器提供可扩展的拆分生成 ，允许查询利用列式存储、谓词下推和元数据修剪来高效地并行处理数十亿条记录。这种转变不仅消除了 PostgreSQL 的可扩展性限制，还显著缩短了查询执行时间，使大规模实时分析成为可能。

在上面的架构图中，可以简单地将第 1 部分工件替换为其他数据库和相应的 CDC 工具——比如我们可以用 MySQL + Debezium 或 Mongodb + Mongo 的 CDC 处理程序等替换 PostgreSQL + Debezium 组合。

围绕 Debezium 构建框架

我们没有为每个数据库 / 表手动设置 Debezium，而是构建了一个配置驱动的框架，使团队能够轻松加入新的数据库 / 表。

我们框架的关键组成部分包括：

• • 连接器管理 — Debezium 连接器的集中配置。
• • Schema Evolution Handling （架构演变处理 ） – 自动将架构传播到数据湖。
• • 监控和警报 — 跟踪失败事件、自动重启、快照、跟踪丢失删除事件等
• • Debezium 信号 — 用于增量快照
• • 心跳查询 — 对于指向在同一 postgres 实例中运行的多个数据库的连接器，在这种情况下，有时 debezium 可能无法从事件频率较低的 db 获取更改。心跳查询可帮助 debezium 确认更改，并将复制槽向前移动。

为什么选择 Apache Hudi 支持数据湖？

虽然 Delta Lake 和 Apache Iceberg 被广泛用于数据湖，但 Apache Hudi 因其能够有效地大规模处理更新插入、增量处理和实时分析而脱颖而出。

Hudi 专为有状态数据湖而构建，针对记录级索引和高效的写入性能进行了优化。与依赖完整文件重写进行更新的 Delta Lake 和 Iceberg 不同，Hudi 采用 Bloom 过滤器、全局索引和存储桶化存储布局来最大限度地减少写入放大，从而大幅减少延迟和存储开销。

其读取时合并 （MOR） 模式通过仅写入增量日志并异步压缩它们来实现低延迟更新，从而在不牺牲查询性能的情况下确保高写入吞吐量。这使得 Hudi 特别适合需要频繁更新大型数据集同时保持快速分析的工作负载。

另一个关键区别是增量查询 。Hudi 允许下游使用者仅获取更改的记录，避免昂贵的全表扫描，这是实时分析、CDC（变更数据捕获）和流式摄取的关键优势。通过利用 Trino 中的结构化分区、元数据修剪和可扩展的拆分生成，Hudi 提供了一种经济高效、高性能的解决方案，用于大规模管理有状态数据湖

挑战和经验教训

在创建和部署 CDC 管道时，我们面临的挑战很少，但我们也解决了

1. 1. 处理 Kafka 集群/PG 切换的迁移 — 由于我们仍在使用 PostgreSQL v15，因此在切换时，Debezium 使用的逻辑复制槽不会传播到备用服务器，因此，我们添加了一个在 PG 切换时拍摄快照的机制，通过删除过时的记录以及一个将在数据湖中保持状态一致的工作流。
2. 2. 拆分大型有效负载 — 在少数情况下，从 postgres 表生成的事件可能具有巨大的 JSON blob，这在开始时阻碍了我们的摄取管道，为了处理此类事件，我们将其拆分为较小的记录块，以实现高效的数据摄取。尽管这也可以在 Producer 级别（即 Debezium）实现
3. 3. 跳过几列/跳过重复事件 — 尽管 debezium 具有跳过特定列集的更改事件的内置功能，并且可以跳过重复事件（UPDATE 事件，而列的值没有任何变化），但我们根据我们的用例在其上整合了自定义逻辑
4. 4. 减少管道中的总体延迟 — 随着我们将大多数以客户为中心的仪表板迁移到基于“Debezium CDC 到 HUDI 摄取”的表，在 PostgreSQL 和数据湖中摄取和维护相同的数据状态至关重要，并且延迟最小，截至目前，我们通过单个 Spark 应用程序跟踪 100 多个表
5. 5. 查询性能优化
   • • 我们在 Hudi 中启用了分区修剪以加快查询速度
   • • 优化了压缩和聚簇策略，以提高读取性能

总结

总结一下最终通过这条管道获得的好处，有些是我们最初打算实现的，还有一些是我们希望实现并最终实现的。

1. 1. 从 PostgreSQL 卸载经常查询的数据： 将 PostgreSQL 数据库的读取 QPS 降低到 10000 QPS，从而提高 PG 可靠性。
2. 2. 利用数据湖进行大规模分析查询： 将分析查询性能提高了** 10 倍**。这显著提高了我们的控制面板性能。
3. 3. 确保事务系统和分析系统之间的数据一致性： 主数据库和 OLAP 存储之间的数据同步延迟小于一分钟。
4. 4. 异步缓存同步： 我们的大多数微服务都利用 Redis 作为读取缓存，从我们的主数据库中检索数据。为了防止多个微服务在缓存失效期间使数据库不堪重负，我们使用基于 Debezium 的 CDC 事件采用了异步缓存同步。这不仅消除了缓存过时问题，还确保了高效的更新传播，显著减少了直接数据库命中并提高了整体系统性能。
5. 5. 有人猜测 HUDI 能够支持适合仪表板的低延迟、实时分析查询工作负载。由于数据可以通过 CDC + HUDI 立即用于分析工作负载，因此一些开始无法通过 RDBMS 的 API 被迁移到 HUDI，而对性能和延迟没有实质性影响。这让功能团队有信心开始放弃 RDBMS 并直接写入 HUDI。
6. 6. 成本优化之路 ： 我们的 OLTP（带副本的 PostgreSQL 或 MongoDB）在成本和运营方面都变得越来越昂贵。他们在繁重的负载下苦苦挣扎，促使我们重新评估 Schema 设计，这项工作将进一步增加开发和 QA 费用。借助 CDC 到 HUDI 摄取管道，我们通过将现有数据库卸载到数据湖，使现有数据库具有一定的寿命，并且如第 4 点所述，我们创建了一个将来扩展的模式。