使用 Apache Hudi 对 Peloton 的数据基础设施进行现代化改造

Peloton 使用 Apache Hudi 重新构建了其数据平台，以克服快照延迟、僵化的服务耦合和高运营成本。通过从 PostgreSQL 和 DynamoDB 采用基于 CDC 的摄取，从 CoW 表迁移到 MoR，并利用具有细粒度架构控制的异步服务，Peloton 实现了 10 分钟的摄取周期，减少了计算/存储开销，并实现了时间旅行和 GDPR 合规性。

Peloton 是一个全球互动健身平台，为全球数百万会员提供互联的、由教练指导的健身体验。Peloton 以其沉浸式课程和尖端设备而闻名，将软件、硬件和数据相结合，创建个性化的锻炼之旅。随着会员群的不断增长和产品多样性的增加，数据已成为 Peloton 如何创造价值的核心。Peloton 的数据_平台_团队负责构建和维护为分析、报告和实时数据应用程序提供支持的核心基础设施。他们的工作确保数据从事务系统无缝流向数据湖，使整个组织的团队能够及时做出数据驱动的决策。

挑战：数据增长、延迟和运营瓶颈

随着 Peloton 发展成为全球交互式健身平台，其数据基础设施面临着对及时洞察、敏捷服务迁移和经济高效的分析日益增长的需求的挑战。日常运营、推荐系统和合规性要求需要一种能够支持近乎实时的访问、高频率更新和可扩展的服务边界的架构。

然而，该团队在现有设置中面临着持续的瓶颈：

• • 报告管道由完整快照作业的完成限制。
• • 推荐系统只能在每日刷新的数据集上运行。
• • 分析平台与作系统紧密结合。
• • 微服务迁移仅限于一次性轮班。
• • 数据库只读副本会产生高昂的基础设施成本。

这些限制使得难以满足 SLA 期望、有效扩展工作负载以及使平台适应新用户和产品需求。

The Legacy Architecture 遗留架构

image.png

Peloton 的早期架构依赖于来自单体 PostgreSQL 数据库的每日快照。分析系统会使用这些快照，通常需要等待数小时才能完成。这不仅延迟了报告，还引入了下游僵化。

由于同一数据平台同时支持联机和分析工作负载，因此任何架构或服务迁移都需要大量的规划和协调。用于缩放读取的数据库只读副本增加了成本开销。此外，依赖于数据新鲜度的推荐系统受到快照间隔的限制，限制了个性化能力。这种架构难以支持快速发展的产品路线图、近乎实时的分析以及实验和迭代所需的数据敏捷性。

使用 Apache Hudi 重新构想数据平台

image.png

为了应对这些挑战，数据平台团队引入了 Apache Hudi 作为其现代数据湖的基础。该架构经过重建，以支持 Debezium 从 PostgreSQL 和 DynamoDB 摄取变更数据捕获 （CDC），Kafka 充当传输层。开发了一个定制的 Hudi 编写器，用于使用 EMR 上的 Apache Spark（版本 6.12.0 和 Hudi 0.13.1）将 CDC 记录摄取到 S3 中。

Peloton 最初选择写入时复制 （CoW） 表格式来支持通过 Redshift Spectrum 进行查询并简化采用。然而，性能和成本瓶颈促使过渡到具有异步表服务的读取合并 （MoR） 表，以进行清理和压缩。

主要架构增强功能包括：

• • 通过结构化删除传播支持 GDPR 合规性 。
• • 用于推荐模型训练和数据恢复的时间旅行查询 。
• • 通过解耦引入对微服务的分阶段迁移支持 。

Peloton 更广泛的数据平台技术堆栈通过一系列用于编排、分析和治理的工具支持这种架构。这包括用于计算的 EMR、用于查询的 Redshift、用于数据转换的 DBT、用于 BI 和可视化的 Looker、用于编排的 Airflow 以及用于元数据管理的 DataHub。这些组件补充了 Apache Hudi，形成了一个模块化和生产就绪的Lakehouse栈。

image.png

从大规模运行 Hudi 中学到的知识

随着 Hudi 现在集成到 Peloton 的数据湖中，该团队开始观察和解决大规模出现的新运营和架构挑战。本部分概述了在保持高引入吞吐量、确保数据可靠性和控制基础结构成本时学到的主要经验教训。

CoW 与 MoR：性能权衡

最初，选择写时复制 （CoW） 表是为了简化部署并确保与 Redshift Spectrum 的兼容性。但是，随着引入频率的增加和更新量跨越数百个分区，性能成为瓶颈。一些跨 256 个分区进行更新的高频表每次运行需要近一个小时才能处理。此外，保留 30 天的提交来训练推荐模型会显着增加存储需求，达到数百 GB。

image.png

为了解决这个问题，该团队迁移到 Hudi 的 Merge-on-Read （MoR） 表，并将提交保留期缩短到 7 天。现在，引入作业每 10 分钟运行一次，延迟显着下降，存储和计算使用变得更加高效。

异步与内联表服务

为了提高写入吞吐量并满足低延迟摄取目标，Peloton 团队最初为 Apache Hudi 配置了异步清理器和压缩器服务。这种方法在大多数表中都运行良好，允许引入管道每 10 分钟运行一次，同时将阻塞降至最低，但引入了一些作边缘情况。遇到的一些挑战包括：

• • 并发执行写入器和清理作业，导致冲突。通过引入基于 DynamoDB 的锁来序列化访问，可以缓解这些问题。
• • 读取器清理器争用条件，其中时间旅行查询间歇性失败，并出现 “找不到文件” 错误 - 可追溯到清理器在读取过程中删除文件。
• • EMR 节点终止导致压缩中断，当作业中途失败时，会导致孤立文件。

这些边缘情况主要是由于在 Peloton 规模下管理并发工作负载的作复杂性。在权衡可靠性和延迟后，该团队选择切换到内联表服务进行压缩和清理，并使用自定义逻辑进行增强，以控制这些作的运行时间。此更改提高了系统稳定性，同时保持了可接受的延迟权衡。

Glue 架构版本限制

随着架构的不断发展，该团队使用 Hudi 的 META_SYNC_ENABLED 将架构更新与 AWS Glue 同步。随着时间的推移，高频模式更新将 Glue 中的 TABLE_VERSION 资源数量推高了 _100 万_个限制。这导致作业以最初难以追踪的方式失败。

其中一个故障表现为以下错误：

ERROR Client: Application diagnostics message: User class threw exception:java.lang.NoSuchMethodError: 'org.apache.hudi.exception.HoodieException org.apache.hudi.sync.common.util.SyncUtilHelpers.getExceptionFromList(java.util.Collection)'

经过大量调试后，该问题被追溯到 AWS Glue 限制。该团队实施了多步骤修复：

• • 与 AWS 合作暂时提高资源限制。
• • 开发了 Python 服务来识别和删除过时的表版本，删除了超过 100 万个条目。
• • 添加了一个 Airflow 作业来安排每周清理任务。
• • 改进了架构同步逻辑，使其仅在架构更改时触发。

Debezium 和 TOAST 处理

由于数据库使用 TOAST（超大属性存储技术）处理大字段，PostgreSQL CDC 摄取带来了独特的挑战。当超过 8KB 的字段保持不变时，Debezium 会 __debezium_unavailable_value 发出占位符值，因此无法确定该值是否发生了变化。

image.png

为了解决这个问题，Peloton：

• • 使用 PostgreSQL 快照填充初始数据。
• • 在传入的 CDC 记录和现有 Hudi 记录之间实现自联接，以填充缺失值。
• • 在 Spark 批处理中分离插入、更新和删除。
• • 使用 ts 字段作为预组合键，以确保仅保留最新的记录状态。

还开发了协调管道来修复因批处理中同一键的多个作（例如，创建-删除-创建）而导致的数据不一致。

数据验证和质量执行

数据质量对于确保对新建立的数据湖的信任至关重要。该团队开发了几个内部库和检查：

• • 一个加密粉碎库，用于在存储前加密 user_id 和其他 PII 字段。
• • 将湖中的记录与快照数据进行比较的数据验证框架。
• • 强制实施列级阈值的数据质量库。这些检查与 DataHub 集成，并与气流传感器相关联，以在出现故障时停止下游作业。

DynamoDB 摄取和架构挑战

一些 Peloton 服务依赖于 DynamoDB 来处理运营工作负载 （NoSQL）。为了将这些数据集摄取到湖中，该团队使用了 DynamoDB Streams 和 Kafka 连接器，从而允许重用现有的基于 Kafka 的 Hudi 摄取路径。

image.png

然而，DynamoDB 的 NoSQL 特性带来了架构管理挑战。评估了两种策略：

1. 1. 利益干系人定义的架构，使用 SUPER 类型字段。
2. 2. 动态模式推理，其中分析传入的 JSON 记录，并推断和协调不断发展的模式。

尽管处理时间增加了，但该团队还是选择了动态推理，因为它可以更好地支持探索性工作负载。每日快照和对帐步骤有助于清理不一致的架构状态。

降低运营成本

image.png

随着系统的成熟，成本优化成为当务之急。该团队使用 Ganglia[1] 分析工作概况并确定需要改进的领域：

• • EMR 资源根据 CPU 和内存使用情况逐渐调整大小。
• • 引入了条件 Hive 同步，以避免每次运行期间不必要的同步作。
• • 发现 Spark 端效率低下，不必要地加载存档的时间线，导致作业花费 4 倍的时间。解决此问题可减少总体延迟和计算资源使用量。

这些作改进显着减少了空闲时间并提高了平台的成本效率。

采用 Hudi 的收益

Peloton 向 Apache Hudi 的过渡使其现代数据平台实现了可衡量的性能、运营和成本相关改进。

Peloton 向 Apache Hudi 的过渡产生了几个可衡量的改进：

• • 摄取频率从每天一次增加到每 10 分钟一次。
• • 快照作业持续时间从一小时缩短到 15 分钟以下。
• • 通过消除只读副本和优化 EMR 集群使用来节省成本。
• • 时间旅行支持支持回顾性和模型重新训练。
• • 通过结构化删除和加密 PII 改善合规性状况。

现代化为未来的发展奠定了基础，包括使用 Apache Flink 进行实时流式摄取，以及数据新鲜度、延迟和治理方面的持续改进。

引用链接

[1] Ganglia: https://github.com/ganglia/