Uber 使用 Apache Hudi 构建了一个大规模事务型数据湖

本文最初发布于 Uber 工程博客，由 InfoQ 中文站翻译并分享。

Apache Hudi 是一个存储抽象框架，帮助分布式组织构建和管理 PB 级的数据湖。Apache Hudi 非常灵活，还可以操作 Hadoop 分布式文件系统（HDFS）或云存储。Hudi 可以在数据湖上提供原子性、一致性、隔离和持久性（ACID）语义。

从准确预测到达时间到预测最优交通路线，在 Uber 平台上提供安全、完美的出行体验需要可靠的、高性能的大规模数据存储和分析。2016 年，Uber 开发了增量处理框架 Apache Hudi，为业务关键数据管道提供低延迟、高效率的支持。一年后，我们选择开源这个解决方案，让其他依赖数据的组织也可以从中获益，然后，在 2019 年，我们更进一步，将其 捐赠 给了 Apache 软件基金会。现在，大约一年半之后，Apache Hudi 已经 成为 Apache 软件基金会的顶级项目 。为了纪念这一里程碑事件，我们想分享 Apache Hudi 构建、发布、优化和毕业的整个过程，希望可以为更广大的大数据社区带来帮助。

Apache Hudi 是什么？

Apache Hudi 是一个存储抽象框架，帮助分布式组织构建和管理 PB 级的数据湖。通过使用诸如 upsert 和 incremental pull 这样的原语，Hudi 将流式处理引入到批处理风格的大数据中。这些特性有助于为我们的服务提供更快、更新的数据，它提供了统一的服务层、分钟级的数据延迟，还避免了维护多个系统的额外开销。Apache Hudi 非常灵活，还可以操作 Hadoop 分布式文件系统（HDFS）或云存储。

Hudi 可以在数据湖上提供原子性、一致性、隔离和持久性（ACID）语义。Hudi 的两个应用最广泛的特性是 upsert 和 incremental pull，它们让用户能够接收变化数据捕获的信息，并将它们大规模地应用到数据湖中。为了实现这一点，Hudi 提供了各种可插拔的索引功能，而且有自己的数据索引实现。Hudi 能够控制和管理数据湖中的文件布局，这非常重要，既摆脱了 HDFS 命名节点和其他云存储的限制，还能通过提高可靠性和查询性能来保证数据生态系统的健康。为此，Hudi 支持多种查询引擎集成，如 Presto、Apache Hive、Apache Spark 和 Apache Impala。

图 1 Apache Hudi 接收变化日志、事件和增量流，暴露表的不同视图来服务于不同的用例。

从高层次上讲，Hudi 在概念上分为 3 个主要组件：需要存储的原始数据、用于提供 upsert 功能的数据索引以及用于管理数据集的元数据。在其内部，Hudi 维护着在不同时间点上执行的所有操作的时间表，在 Hudi 中称为 instants。这让它可以提供表的即时视图，同时还支持高效地按到达顺序检索数据。根据时间表上的时刻（换句话说，在数据库中进行更改的时间），Hudi 可以保证操作是原子的、一致的。有了这些信息，Hudi 就可以提供同一个 Hudi 表的不同视图，包括具有高效列查询性能的读优化视图、用于快速数据摄入的实时视图和将 Hudi 表作为更改日志流读取的增量视图，如图 1 所示。

Hudi 将数据表组织到分布式文件系统中的一个基路径下。表分为多个分区，每个分区内，文件被组织成文件组，每个组有惟一的文件 ID 标识。每个文件组包含几个文件片，每个文件片包含一个基文件（* .parquet），这个文件是在 commit/compaction 时生成的，同时生成的还有日志文件（* . log . *），其中包含自基文件创建以来的插入 / 更新操作。Hudi 采用多版本并发控制（MVCC），其中压缩操作会合并日志和基文件，生成新的文件片，清理操作会处理掉未使用的 / 旧的文件片，从而回收文件系统上的空间。

Hudi 支持两种表类型：写时复制和读时合并。写时复制表使用列文件格式（例如，Apache Parquet）存储数据。写时复制会更新文件的版本，并通过写时同步合并来重写文件。

读取合并表综合使用了基于列（例如 Apache parquet）和行（例如 Apache Avro）的文件格式来存储数据。更新被记录到增量文件中，然后再压缩，同步或异步地生成新版本的列文件。

Hudi 还支持两种查询类型：快照查询和增量查询。快照查询需要在特定提交或压缩操作时生成表的“快照”。在利用快照查询时，写时复制表只暴露最新文件片中的基 / 列文件，并保证与非 Hudi 列表相同的列查询性能。写时复制提供了替换现有 Parquet 表的功能，同时提供了更新 / 删除及其他功能。对于读时合并表，快照查询通过动态合并最新文件片的基文件和增量文件来暴露近实时数据（以分钟为单位）。对于写时复制表，增量查询提供在特定提交或压缩之后写入到表中的新数据，从而可以提供更改流以支持增量数据管道。

Apache Hudi 在 Uber 的使用

在 Uber，我们在各种用例中使用了 Hudi，从在 Uber 平台上提供快速、准确的出行数据，到 检测欺诈行为 ，再到 在UberEats 平台上推荐餐厅和美食 。为了演示Hudi 的工作机制，让我们来看看，我们 如何确保Uber 市场上的出行数据在数据湖上是最新的 ，从而提升Uber 平台上乘客和司机的用户体验。旅程的典型生命周期是从乘客请求开始，随着旅程的进行而继续，并在旅程结束且乘客到达最终目的地时结束。Uber 的核心出行数据以表的形式存储在 Schemaless （Uber 的可伸缩数据存储）中。 在旅程的生命周期中，旅程表中的单个旅程条目可能会经历多次更新。在 Uber 实现 Hudi 之前，一些大型的 Apache Spark 作业会定期将整个数据集重写到 Apache HDFS，合并上游在线表的插入、更新和删除，反映旅程状态的变化。在 2016 年初（在我们建立 Hudi 之前），我们一些最大的任务使用了超过 1000 个执行器，处理超过 20TB 的数据。这个过程不仅效率不高，而且也难以扩展。公司里的各个团队都依赖快速、准确的数据分析来提供高质量的用户体验，很明显，我们目前的解决方案无法满足需求，它无法通过扩展实现面向数据湖的便捷的增量处理。在使用快照和重载解决方案将数据移动到 HDFS 时，这种低效率会影响到所有管道，包括使用原始数据的下游 ETL。我们看到，随着规模的扩大，这些问题只会越来越严重。

在没有其他可行的开源解决方案可供我们使用的情况下，我们在 2016 年底 构建并推出了 Hudi ，用于构建一个事务型数据湖，实现快速、可靠的大规模数据更新。在 Uber，第一代 Hudi 仅使用了写时复制表，就将作业处理速度提高到每 30 分钟 20GB，将 I/O 和写放大率减少了 100 倍。到 2017 年底，Uber 的所有原始数据表都采用了 Hudi 格式，我们的数据湖是全球最大的事务型数据湖之一。

图 2 Hudi 的“写时复制”功能使我们能够执行文件级的更新，大幅提高了数据新鲜度。

改进 Apache Hudi，不只为 Uber

随着 Uber 数据处理和存储需求的增长，Hudi 的写时复制功能开始显出局限性，主要是因为我们需要继续提高数据的呈现速度和新鲜度。即使使用了 Hudi 的写时复制功能，我们的一些表收到的更新也覆盖了 90% 的文件，导致数据湖中每个大型表的数据重写量都在 100TB 左右。因为即使只修改一条记录，写时复制功能也会重写整个文件，使写放大率增加而数据新鲜度降低，导致 HDFS 集群上不必要的 I/O，加速磁盘性能退化。此外，更多的数据表更新意味着更多的文件版本以及 HDFS 文件数量的爆炸。反过来，这会导致 HDFS 命名节点不稳定，计算成本增加。

为了解决这些日益增多的问题，我们实现了第二种表类型，即读取合并。由于读取合并通过动态合并数据来使用近实时数据，所以谨慎起见，我们使用该特性来避免查询端计算成本。我们的读取合并部署模型包括三个独立的作业：一个摄入作业，带来由插入、更新和删除组成的新的数据；一个次要压缩作业，在少量最新的分区中积极地对更新 / 删除进行异步压缩；一个主要压缩作业，在大量的旧分区中缓慢而稳定地压缩更新 / 删除。每个作业都以不同的频率运行，次要作业和摄入作业比主要作业运行得更频繁，从而确保最新分区中的数据能以列格式快速提供。有了这样一个部署模型，我们就能够以列格式向数千个查询提供新数据，并将查询端合并成本限定在最新的分区上。使用读取合并，我们能够解决上面提到的所有三个问题，并且，不管数据湖更新或删除的量多大，Hudi 表都几乎不受影响。现在，在 Uber，我们根据情况同时使用了 Apache Hudi 的写时复制和读时合并功能。

图 3 Uber 的 Apache Hudi 团队为读取合并表开发了一种数据压缩策略，以便频繁地将最新分区转换为列格式，从而减少查询端计算成本。

借助 Hudi，Uber 每天将超过 5000 亿条的记录接收到我们 150PB 的数据湖里，跨 10000 多个表和数千条数据管道，每天使用超过 30000 个虚拟内核。Hudi 表每周要为我们的各种服务提供超过 100 万次查询服务。

回顾

Uber 在 2017 年将 Hudi 开源，让其他人可以受益于这样一个大规模摄入和管理数据存储的解决方案，将流处理引入大数据。随着 Hudi 毕业成为 Apache 软件基金会的顶级项目，Uber 的大数据团队回顾了当初激励我们创建 Hudi 的各种考虑，包括：

• 我们的数据存储和处理如何才能更高效？
  • 我们如何确保我们的数据湖包含高质量的表？
  • 随着运营规模的增长，我们如何继续以较低的延迟高效地提供数据？
  • 如何为可以延迟几分钟的用例提供统一的服务层？

如果没有良好的标准化和原语，数据湖很快就会变成无法使用的“数据沼泽”。这样的话，不仅核对、清理和修复表需要大量的时间和资源，而且各个服务的所有者还不得不为调优、重排和事务构建复杂的算法，给技术堆栈带来不必要的复杂性。

如上所述，Hudi 通过在分布式文件系统上无缝地摄入和管理大型分析型数据集来帮助用户控制他们的数据湖，从而解决了这些问题。构建数据湖涉及多方面的问题，需要投入时间和精力来研究数据标准化、存储技术、文件管理实践、在数据摄入和数据查询之间做恰当的性能取舍等。在构建 Hudi 的过程中，我们与大数据社区的其他成员进行了交流，我们了解到，这样的问题在许多工程组织中都很普遍。我们希望，在过去的几年里，开源以及与 Apache 社区合作进行的以 Hudi 为基础的构建，能够让其他人更深入地了解他们自己的大数据运营，从而改进各行业的应用程序。除了 Uber，还有 多家公司 在生产环境中使用了 Apache Hudi，其中包括阿里云、Udemy 和腾讯。

展望

图 4 Apache Hudi 的用例包括数据分析和基础设施健康监控。

Hudi 帮助用户构建更健壮、新鲜度更高的数据湖，通过数据集规范化提供高质量的洞察。

Uber 拥有世界上最大的事务型数据湖之一，这让我们有机会找到独特的 Apache Hudi 用例。鉴于在这种规模下解决问题和提升效率会产生重大的影响，我们就有了更深入研究的直接动机。在 Uber，我们已经使用高级的 Hudi 原语（如 incremental pull ）来帮助我们构建链式增量管道，减少作业占用的计算资源，不然的话，这些作业将执行大规模扫描和写入操作。我们根据自己特定的用例和需求，对读取合并表的 压缩 策略进行了优化。在最近几个月里，我们将 Hudi 捐赠给了 Apache 基金会，并贡献了多项功能，比如， 支持高效文件系统访问的嵌入式时间表服务 ， 删除重命名以支持云友好的部署 ，以及 提高 incremental pull 性能 等等。

在接下来的几个月里，Uber 打算为 Apache Hudi 社区贡献多项新特性。其中一些特性通过优化计算资源使用以及提高数据应用程序的性能来帮助降低成本。我们还将深入研究如何根据访问模式和数据应用程序需求改进存储管理和查询性能。

要了解关于我们计划如何实现这些目标的更多信息，可以阅读一些由 Hudi 团队提出、旨在扩展 Apache 社区的 RFC（包括 用于支持列索引和 O(1) 查询计划的智能元数据 ， 将表高效迁移到 Hudi ， 用于快速 upsert 的记录级索引 ）。

Apache Hudi 已经毕业成为顶级 Apache 项目，我们很高兴能为这个项目的宏伟蓝图做出贡献。Hudi 让 Uber 和其他公司能够提升数据湖的速度、可靠性和事务能力，利用开源文件格式，将许多大数据的挑战抽象出来，构建丰富的、可移植的数据应用程序。

Apache Hudi 是一个正在成长的社区，拥有一个令人兴奋的、不断发展的开发路线图。如果你有兴趣为这个项目贡献，请点击 这里 。

关于作者

Nishith Agarwal 是 Uber 和 Apache Hudi PMC 大数据平台的工程负责人。他的团队帮助 Uber 构建数据湖基础设施。

查看英文原文： Building a Large-scale Transactional Data Lake at Uber Using Apache Hudi