Apache Pulsar结合Hudi构建Lakehouse

1. 动机

Lakehouse最早由Databricks公司提出，其可作为低成本、直接访问云存储并提供传统DBMS管系统性能和ACID事务、版本、审计、索引、缓存、查询优化的数据管理系统，Lakehouse结合数据湖和数据仓库的优点：包括数据湖的低成本存储和开放数据格式访问，数据仓库强大的管理和优化能力。Delta Lake，Apache Hudi和Apache Iceberg是三种构建Lakehouse的技术。

与此同时，Pulsar提供了一系列特性：包括分层存储、流式卸载、列式卸载等，让其成为一个可以统一批和事件流的存储层。特别是分层存储的特性，然Pulsar成为一个轻量级数据湖，但是Pulsar还是缺乏一些性能优化，比如索引，数据版本（在传统DBMS管理系统中非常常见），引入列式卸载程序的目的是为了缩小性能差距，但是还不够。

本提议尝试将Apache Pulsar作为Lakehouse，该提案仅提供顶层设计，详细设计和实现在后面的子提议中解决；

2. 分析

本部分将分析构建Lakehouse需要的关键特性，然后分析Pulsar是否满足要求以及识别还有哪些差距。

Lakehouse有如下关键特性：

• 事务支持：企业级Lakehouse中很多数据pipeliine会并发读写数据，支持ACID事务可以保证并发读写的一致性，特别是使用SQL；Delta Lake，Iceberg，Hudi三个数据湖框架都基于低成本的对象存储实现了事务层，都支持事务。Pulsar在2.7.0版本后引入了事务支持，并且支持跨topic的事务；
• Schema约束和治理：Lakehouse需要支持Schema的约束和演进，支持数仓型Schema范式，如星型/雪花型Schema，另外系统应该能够推理数据完整性，并且应该具有健壮的治理和审核机制，上述三个系统都有该能力。Pulsar有内置的Schema注册服务，它满足Schema约束和治理的基本要求，但是可能仍有一些地方需要改进。
• BI支持：Lakehouses可以直接在源数据上使用BI工具，这样可以减少陈旧性，提高新鲜度，减少等待时间，并降低必须同时在数据湖和仓库中操作两个数据副本的成本。三个数据湖框架与Apache Spark的集成非常好，同时可以允许Redshift，Presto/Athena查询源数据，Hudi社区也已经完成了对多引擎如Flink的支持。Pulsar暴露了分层存储中的段以供直接访问，这样可以与流行的数据处理引擎紧密集成。 但是Pulsar中的分层存储本身在服务BI工作负载方面仍然存在性能差距，我们将在该提案中解决这些差距。
• 存储与计算分离：这意味着存储和计算使用单独的集群，因此这些系统可以单独水平无限扩容。三个框均支持存储与计算分离。Pulsar使用了存储与计算分离的多层体系结构部署。
• 开放性：使用开放和标准化的数据格式，如Parquet，并且它们提供了API，因此各种工具和引擎（包括机器学习和Python / R库）可以"直接"有效地访问数据，三个框架支持Parquet格式，Iceberg还支持ORC格式，对于ORC格式Hudi社区正在支持中。Pulsar还不支持任何开放格式，列存卸载支持Parquet格式。
• 支持从非结构化数据到结构化数据的多种数据类型：Lakehouse可用于存储，优化，分析和访问许多新数据应用程序所需的数据类型，包括图像，视频，音频，半结构化数据和文本。尚不清楚Delta，Iceberg，Hudi如何支持这一点。Pulsar支持各种类型数据。
• 支持各种工作负载：包括数据科学，机器学习以及SQL和分析。 可能需要多种工具来支持所有这些工作负载，但它们都依赖于同一数据存储库。三个框架与Spark紧密结合，Spark提供了广泛的工具选择。Pulsar也与Spark有着紧密结合。
• 端到端流：实时报告是许多企业的常态，对流的支持消除了对专门用于服务实时数据应用程序的单独系统的需求，Delta Lake和Hudi通过变更日志提供了流功能。 但这不是真正的“流”。Pulsar是一个真正的流系统。

可以看到Pulsar满足构建Lakehouse的所有条件。然而现在的分层存储有很大的性能差距，例如：

• Pulsar并不以开放和标准的格式存储数据，如Parquet；
• Pulsar不会为卸载的数据部署任何索引机制；
• Plusar不支持高效的Upserts；

这里旨在解决Pulsar存储层的性能问题，使Pulsar能作为Lakehouse。

3. 当前方案

图1展示了当前Pulsar流的存储布局。

• Pulsar在ZooKeeper中存储了段（segment）元数据；
• 最新的段存储在Apache BookKeeper中（更快地存储层）
• 旧的段从Apache BookKeeper卸载到分层存储（便宜的存储层）。 卸载的段的元数据仍保留在Zookeeper中，引用的是分层存储中卸载的对象。

当前的方案有一些缺点：

• 它不使用任何开放式存储格式来存储卸载的数据。 这意味着很难与更广泛的生态系统整合。
• 它将所有元数据信息保留在ZooKeeper中，这可能会限制可伸缩性。

4. 新的Lakehouse存储方案

新方案建议在分层存储中使用Lakehouse存储卸载的数据。该提案建议使用Apache Hudi作为Lakehouse存储，原因如下：

• 云提供商在Apache Hudi上提供了很好的支持。
• Apache Hudi已经作为顶级项目毕业。
• Apache Hudi同时支持Spark和Flink多引擎。同时在中国有一个相当活跃的社区。

4.1 新的存储布局

图2展示了Pulsar topic新的布局。

• 最新片段（未卸载片段）的元数据存储在ZooKeeper中。
• 最新片段（未卸载片段）的数据存储在BookKeeper中。
• 卸载段的元数据和数据直接存储在分层存储中。 因为它是仅追加流。 我们不必使用像Apache Hudi这样的Lakehouse存储库。 但是如果我们也将元数据存储在分层存储中，则使用Lakehouse存储库来确保ACID更有意义。

4.2 支持高效Upserts

Pulsar不直接支持upsert。它通过主题（topic）压缩支持upsert。 但是当前的主题压缩方法既不可扩展，也不高效。

• 主题压缩在代理内（broker）完成。 它无法支持大量数据的插入，特别是在数据集很大的情况下。
• 主题压缩不支持将数据存储在分层存储中。

为了支持高效且可扩展的Upsert，该提案建议使用Apache Hudi将压缩后的数据存储在分层存储中。 图3展示了使用Apache Hudi支持主题压缩中的有效upserts的方法。

该想法是实现主题压缩服务。主题压缩服务可以作为单独的服务（即Pulsar函数）运行以压缩主题。

• 代理向压缩服务发出主题压缩请求。
• 压缩服务接收压缩请求，并读取消息并将其向上插入到Hudi表中。
• 完成upsert之后，将主题压缩游标前进到它压缩的最后一条消息。

主题压缩游标将引用位置的元数据存储在存储Hudi表的分层存储中。

4.3 将Hudi表当做Pulsar Topic

Hudi会在不同的即时时间维护对表执行的所有操作的时间轴，这有助于提供表的即时视图，同时还有效地支持按_arrival_顺序进行数据检索。Hudi支持从表中增量拉取变更。我们可以支持通过Hudi表备份的_ReadOnly_主题。这允许应用程序从Pulsar代理流式传输Hudi表的变更。图4展示了这个想法。

4.4 可扩展的元数据管理

当我们开始将所有数据存储在分层存储中时，该提案建议不存储卸载或压缩数据的元数据，而只依赖分层存储来存储卸载或压缩数据的元数据。

该提案提议在以下目录布局中组织卸载和压缩的数据。

- <tenant>/
  - <namespace>/
    - <topics>/
      - segments/ <= Use Hudi to store the list of segments to guarantee ACID
        - segment_<segment-id>
        - ...
      - cursors/
        - <cursor A>/ <= Use Hudi to store the compacted table for cursor A.
        - <cursor B>/ <= ...

5. 引用

[1] Lakehouse: A New Generation of Open Platforms that Unify Data Warehousing and Advanced Analytics. http://cidrdb.org/cidr2021/papers/cidr2021_paper17.pdf

[2] What is a Lakehouse? https://databricks.com/blog/2020/01/30/what-is-a-data-lakehouse.html

[3] Diving Deep into the inner workings of the Lakehouse and Delta Lake. https://databricks.com/blog/2020/09/10/diving-deep-into-the-inner-workings-of-the-lakehouse-and-delta-lake.html