Robinhood基于Apache

1. 摘要

Robinhood 的使命是使所有人的金融民主化。 Robinhood 内部不同级别的持续数据分析和数据驱动决策是实现这一使命的基础。 我们有各种数据源——OLTP 数据库、事件流和各种第 3 方数据源。需要快速、可靠、安全和以隐私为中心的数据湖摄取服务来支持各种报告、关键业务管道和仪表板。
不仅在数据存储规模和查询方面，也在我们在数据湖支持的用例方面，我们从最初的数据湖版本都取得了很大的进展。在这篇博客中，我们将描述如何使用各种开源工具构建基于变更数据捕获的增量摄取，以将我们核心数据集的数据新鲜延迟从 1 天减少到 15 分钟以下。 我们还将描述大批量摄取模型中的局限性，以及在大规模操作增量摄取管道时学到的经验教训。

2. 数据湖和生态系统

Robinhood 的数据湖存储和计算基础架构是为我们的许多数据驱动功能提供支持的基石，例如业务分析仪表板和产品改进见解。 它也是为业务和临时报告和分析运行大规模数据处理的数据源。 此外，生态系统会影响以隐私为中心的原语，例如旨在保护用户隐私的匿名化和访问控制。

主要的 OLTP（在线事务处理）数据库由 Postgres RDS 管理；Amazon S3 是 Data Lake 存储，它为我们的 Data Lake 提供经济高效且可扩展的存储层；我们主要使用 Apache Spark 运行生产批处理管道；我们的仪表板由 Trino 分布式 SQL 查询引擎提供支持；Apache Hadoop Yarn 管理用于运行 Apache Spark 作业的计算集群；Apache Hive Metastore 为查询引擎管理和提供表模式；Apache Airflow 是工作流编排服务。
下图是具有计算生态系统的数据湖

在整篇文章中我们使用指标“数据新鲜度”来比较下面不同的数据摄取架构，此指标为源数据库中的表中发生的更改在相应的 Data Lake 表中可见提供了时间延迟。

3. 大批量摄取的限制

作为数据湖演进的第一步，我们首先使用在线数据库的只读副本获取在线数据库的每日快照。 摄取这些表的完整快照会导致数据湖表的写入放大率很高。 即使对于一个有数十亿行的表来说，一天只有几十万行的变化，摄取该表的完整快照也会导致读取和写入整个表。 此外当使用实时副本（而不是作为上游的数据库备份）时，在只读副本 I/O 性能方面会出现瓶颈，这会导致快照时间过长，从而导致较大的摄取延迟。 即使采用了诸如通过分区读取并行化 I/O 之类的技术，这种摄取架构也无法在一小时内交付数据。
Robinhood 确实需要保持数据湖的低数据新鲜度。 许多过去在市场交易时间之后或之前以每日节奏运行的批处理管道必须以每小时或更高的频率运行，以支持不断发展的用例。 很明显我们需要更快的摄取管道将在线数据库复制到数据湖。

4. 新架构

实现 Data Lake 较低数据新鲜度的更好方法是增量摄取。 增量摄取是一种众所周知的技术，用于为数据湖构建有效的摄取管道。 在这里摄取管道不是拍摄快照并将它们作为一个整体转储到 Data Lake，而是以流方式使用 OLTP 数据库的预写日志并将它们摄取到 Data Lake 表中，就像数据库到数据库复制的方式一样。
从概念上讲，我们有一个两阶段管道。

• 变更数据捕获 (CDC) 服务使用 OLTP 数据库中的预写日志 (WAL) 数据并将它们缓冲在变更日志队列中。


• 数据摄取作业定期或以连续方式拖尾队列并更新数据湖“原始”表。

下图是增量摄取组件

中间更改日志队列允许分离两个阶段之间的关注点，这两个阶段将能够独立运行，并且每个阶段都可以暂停而不影响另一个阶段。 队列提供了必要的隔离，以便将数据摄取到数据湖的任何延迟都不会对 CDC 造成背压。
在第一阶段，我们选择 Debezium 作为变更数据捕获 (CDC) 提供商。 Debezium 是一个构建在 Kafka Connect 之上的开源分布式变更数据捕获平台，Debezium 带有一个经过充分证明的一流 Postgres CDC 连接器。根据我们的基准测试，我们发现 Debezium 可以轻松处理我们预计的负载量，我们已经设置 Debezium 使用开源的 Confluent Schema Registry 以 avro 编码格式将更改记录写入 Kafka，与 json 编码相比，Avro 编码提供了更好的性能。
在第二阶段，我们使用 Apache Hudi 从 Kafka 增量摄取变更日志，以创建数据湖表。 Apache Hudi 是一个统一的数据湖平台，用于在数据湖上执行批处理和流处理，Apache Hudi 带有一个功能齐全的基于 Spark 的开箱即用的摄取系统，称为 Deltastreamer，具有一流的 Kafka 集成和一次性写入功能，与不可变数据不同，我们的 CDC 数据有相当大比例的更新和删除，Hudi Deltastreamer 利用其可插入的记录级索引在 Data Lake 表上执行快速高效的 upserts，Hudi 通过自动清理旧文件版本、数据Clustering、Hive表模式同步和文件大小调整来自我管理其表，以写入大小合适的文件，原始表当前以 Hudi 的写时复制模式存储，该模式提供原生列式读取性能。

5. 效果总结

我们已经部署了增量摄取管道，以将 1000 个 Postgres 表摄取到数据湖中。在新架构之前，由于快照的限制和所涉及的成本，这些表只能保证能够以每天的节奏进行快照。 使用这种新架构，Data Lake 用户很高兴看到关键表的数据新鲜度从 24 小时缩短到 15 分钟以下。
大批量快照运行时间显示快照表的运行时间长。 请注意由于只读副本 I/O 瓶颈，其中许多表的快照需要按顺序运行。

显示大批量快照的大批量快照运行计划每天仅运行一次，这是因为从数据库中快照所有表的周转时间很长。

新的增量摄取数据新鲜度显示新摄取系统的端到端数据新鲜度约为 5 分钟。​

6. 经验教训

在本节中我们将分享在大规模构建增量摄取管道时学到的经验教训。 我们希望这对任何希望为他们的数据湖踏上类似旅程的人来说都是有价值的。

7. 可缩放的初始引导程序

对数据湖的增量摄取仍然需要源表的初始快照。 Debezium 确实提供了初始快照模式，但需要查询主 RDS 实例，我们不想查询主 RDS 实例以进行快照，以避免生产 OLTP 查询与初始快照查询之间的任何资源竞争。此外，我们需要通过以无锁方式运行并发分区查询以及从数据库备份中获取快照来优化初始快照时间的能力。
出于这些原因，我们在 Apache Hudi Deltastreamer 之上提供了专用的只读副本并实现了一个自定义快照器，它利用 Spark 运行并发分区快照查询来获取表的初始快照，Apache Hudi 的可插拔源框架允许我们用几行代码无缝实现这一点。
对于带外初始快照，我们需要在增量摄取和快照之间切换时仔细跟踪 CDC 流中的正确水印，使用 Kafka，数据摄取作业的 CDC 水印转换为 Kafka 偏移量，这标志着要应用于快照表的开始更改日志事件，如果我们选择一个任意的 Kafka 偏移量，我们最终可能会错过一些应用到 Data Lake 表的更改事件。
​

从概念上讲，我们需要 3 个阶段来执行正确的快照并过渡到增量摄取：

• 保存最新的 Kafka 偏移量，以在切换到增量摄取时用于重播变更日志。 设“Tₛ”为最新事件的源时间。


• 确保只读副本在时间“Tₛ + Δ”时是最新的，其中 Δ 表示捕获 kafka 偏移量以及额外缓冲时间时的 Debezium 延迟。 否则，整个方程式将无法保证 0% 的数据丢失。 从只读副本中获取表的初始快照并创建 Data Lake 表


• 从之前存储的 kafka 偏移量开始消费并执行表的增量摄取。 一旦增量摄取开始发生，将配置单元表定义同步到数据的最新位置，下游消费者现在将能够查询新引导的表。

下图是使用引导架构的增量摄取架构

从专用只读副本进行快照具有局限性，例如副本端的 I/O 瓶颈以及 24 * 7 在线维护只读副本的成本开销。 我们正在探索一种对 OLTP 数据库进行按需备份并使用 AWS S3 导出发布到 S3 的方法。 然后我们可以依靠大规模处理这些 S3 导出并构建初始快照，这种机制可能允许更快的快照并克服只读副本端的一些 I/O 瓶颈。

8. 使用 Postgres 逻辑复制监控背压风险

Postgres 逻辑复制需要 CDC 连接器直连主 RDS。Postgres 逻辑复制协议保证保留 WAL 日志文件，直到 Debezium 完全处理它们。
如果 Debezium 卡住或无法跟上消耗 WAL 日志的速度，这可能会导致 WAL 日志文件累积并耗尽可用磁盘空间，Debezium 社区建议密切监视滞后消息，我们的 Debezium 负载测试也让我们对 Debezium 能够处理预计的变更速度增加充满信心。

9. 自动化恢复

从每日快照切换到增量摄取的副作用之一是摄取工作流变得有状态。管道可能处于快照或增量摄取状态。
此外，还需要执行架构升级、监控和数据质量验证等其他操作，新表和数据库需要定期地加入。
端到端管道涉及不同的系统——在线 CDC 世界和数据湖的批处理/流摄取。为 1000 个表执行入职和常规操作需要适当的状态管理和自动化。
我们意识到我们需要在内部构建一流的编排服务，该服务将利用 Apache Airflow 来管理摄取管道、跟踪载入和表状态并自动处理状态转换和其他维护，这有助于我们大规模运营管道。

10. 并非所有表都是平等的

当谈到这些表对我们的关键用例的重要性时，pareto原则是有效的，我们有一小部分关键表需要在 15 分钟内保证数据新鲜度，我们采取了一种方法，根据表的重要性将表分类为不同的层，高度关键的表被标记为第 0 层，对于这些表，我们提供了一个单独的 CDC 复制槽，以将这些关键表的 CDC 通道与其他表的通道隔离。 此外我们为 Hudi deltastreamer 提供了专门的资源，以持续摄取增量更改日志，并能够在 5 -15 分钟内保持数据最新。 对于较低优先级的表，Hudi deltastreamer 配置为以批处理模式每 15 分钟运行一次。

11. 管理 Postgres 模式更新

我们的业务是将表从在线 OLTP 世界复制到 Data Lake 世界，复制的数据不是不透明的，而是具有适当的模式，并且复制管道保证了将在线表模式转换为数据湖的模式的明确定义的行为。
鉴于 Data Lakes 还能够存储数据更改的整个历史，因此在线和 Data Lake 世界的向后兼容性意味着什么不同。例如，在在线世界中，向 postgres 添加一个不可为空的列是非常好的，但不会遵守用于存储动态变更日志的 Avro（或 Protobuf）的模式演变规则。拥有明确定义的架构演化合约有助于保持数据湖管道更加稳定。
我们发现大多数时候，Schema更改涉及添加新列，我们正在使用 Debezium 功能来冻结我们从 Postgres 表中读取的列集，并依靠重新引导表来处理模式升级，我们计划为端到端管道添加模式兼容性检测机制，以减少重新引导的次数。

12. 未来规划

我们看到使用增量摄取的原始数据湖表的采用速度更快，并且我们正在不断努力提高管道的可靠性。以下是我们正在着手的一些后续步骤：

• 数据质量保证：我们实施了以不同频率运行的通用和自定义数据质量和完整性检查，以发现复制数据中的差异，我们正在努力利用 Apache Hudi 的预提交验证支持在每批提交之前运行自定义验证。


• 进一步减少数据新鲜度滞后：我们目前使用的是 Apache Hudi Copy-On-Write 格式。在这种模式下，我们可以看到大约 5-15 分钟范围内的数据新鲜度，我们计划探索 Apache Hudi 的 Merge-On-Read 格式，以进一步降低数据新鲜度。


• 流式数据湖：Apache Hudi 提供增量处理能力，就像数据库变更日志一样，我们未来的工作涉及使用这种原语并构建端到端流管道以有效地将更改渗透到下游表，这也将使我们能够以实时流媒体的方式执行隐私保护操作，例如屏蔽和匿名化。


• 用于服务间数据交换的 CDC 服务：CDC 已在 Robinhood 中用于为数据湖的增量摄取提供更改流，我们正在研究使用 CDC 流在各种在线微服务之间进行可靠的数据交换。


• 数据计算：我们一直致力于提高基于 Apache Spark 和 Trino 构建的数据计算平台的可用性、效率和性能，以支持关键数据计算工作负载。

这些是在 Robinhood 数据基础设施团队工作的激动人心的时刻，因为我们已经开始构建下一代 Robinhood 数据湖。