Shared Foundations: Modernizing Meta’s Data Lakehouse

technology
, , ,

What Is Shared Foundations

Shared Foundations（共享基础设施）是Meta的一个跨组织的工程，涉及数十个工程团队，推广基于可重用组件、整合引擎、通用API和一致标准原则的组合方法。通过整合计算引擎、汇聚存储库和元数据，以及统一SQL方言和执行引擎，创造了一个更现代化的数据架构；从而提供更好性能、更丰富功能、更高工程速度和更一致用户体验的架构。最终，这使得Meta的数据湖仓库技术栈更能适应当前和未来的趋势，促进了更快的创新。

Why Shared Foundations

业界趋势

在过去的十年里，大规模数据湖仓库的需求发生了演变。除了由新产品、日益复杂的需求和纯粹的有机增长推动的指数级数据增长外，对更新鲜数据和更快查询的需求也在增加，这对降低洞察时间至关重要。数据模型变得更加复杂，导致大多数表包含诸如structs、maps和arrays等复杂数据类型。与此同时，查询复杂性也在增长；越来越常见的是具有大量阶段、迭代图查询、时间序列分析和复杂业务逻辑的查询，尤其是在数据管道中。

硬件和软件的其他趋势也影响了现代数据湖仓库的架构，如利用SIMD指令、GPU和其他与软件协同优化的专用硬件进行本地代码优化。随着网络速度更快和存储单元更大，存储计算分离变得普遍，成为现代数据仓库的首选架构。最后，机器学习工作负载的最近出现引发了一系列新的趋势，包括数据量、复杂性和不寻常的访问模式。

Meta现状

Meta的数仓是基于Hive构建的，它的几个设计原则至今仍然大部分适用：

  • 数据，元数据和计算分离，允许独立扩缩容。
  • 数据存储在HDFS，现在是Meta的下一代分布式文件系统Tectonic中，可以独立于计算横向扩容。
  • 元数据存储在MySQL中，Hive Metastore提供分区机制以优化数据组织，并增加了对分片MySQL的存储支持。传统上，分片是基于每个命名空间进行的。
  • 数据格式是列式的。原始的列式格式RCFile被改进为ORC，如多种编码、NULL支持、灵活的压缩级别以及对arrays和maps等复杂类型的支持，现在使用ORC的内部变体DWRF，它具有更好地支持大型map和加密等附加功能。
  • 由于计算和存储是分离的，多个引擎可以在Hive仓库上运行，如Spark（取代了原始的Hive引擎）、Presto等。

但是Meta的基于Hive的数仓架构也有其限制：

  • 没有流处理的支持。这导致了多年来建立了各种流处理系统，尤其是Puma，这些系统通常与Hive集成得不太好。
  • Hive不支持实时数据摄入。这导致了像Scuba [1]这样的系统的出现，它最初是为了大规模日志分析而构建的，因此缺乏对准确结果或复杂查询的支持，被用于分析。Scuba用C++编写，有自己的SQL方言，并使用自己的文件格式和本地存储，而不是Hive。
  • 编程语言的分歧。大部分DI技术栈（Hive、运行在其上的引擎、Puma等）都是用Java编写的；然而，Meta的大部分使用C++，例如机器学习系统、Tectonic、RocksDB等。Java在Meta内部也不是主要支持的语言。这导致了新引擎如Scuba和Cubrick [20]需要用C++重建许多核心组件，如执行原语（例如函数和操作符）、用于读写ORC等数据格式的编解码器，甚至开发自己的文件格式。可重用组件和编程语言收敛的缺乏导致了重复实现和增加了维护负担。
  • 查询延迟高。Hive引擎对于交互式分析来说太慢了。这导致了多个新引擎如Presto、Scuba、Raptor和Cubrick试图解决交互式分析领域的问题。有些是用Java编写的，有些是用C++编写的，导致了碎片化。此外，即使是用相同语言编写的引擎（例如Scuba和Cubrick，或Presto和Spark），也由于各种历史原因而没有共享任何代码或组件。一个特别典型的例子是，Spark和Presto都读写存储在Hive上的相同ORC格式，但使用完全不同的库与文件系统和元数据存储进行交互，编码和解码ORC文件，甚至解压缩数据！类似的模式也出现在流处理和批处理领域。
  • I/O使用效率低。Hive数据传统上存储在HDFS（后来是Tectonic）存储节点的硬盘中。从硬盘上通过网络获取数据对于许多交互式分析场景来说太慢了。因此，许多交互式引擎（Raptor、Cubrick、Scuba）都设计为计算和存储共同定位，数据需要预先加载到本地SSD或内存中进行查询。这导致了进一步的存储碎片化和数据重复，以及资源闲置。

Meta的数据技术栈在过去的十年里只是逐步演变。这导致了一个难以维护和发展的碎片化技术栈，由近十种SQL方言、针对类似工作负载的多个引擎（每个引擎都有自己的特点）以及不同位置和格式的相同数据的多个副本组成。缺乏标准化和可重用组件不仅增加了工程团队的操作负担，而且最终减缓了创新的速度。这也影响了Meta的用户，他们不得不与暴露不同SQL方言、不一致语义并呈现次优性能的引擎进行交互。

How Shared Foundations

Shared Foundations基于如下核心设计原则：

  • 更少的系统，比如，交互式、批处理、流处理和机器学习领域应该只需要存在一个单一的计算引擎，提供比合并前的碎片化系统更多的功能。在此过程中，还应删除任何剩余数据或重复的管道。
  • 共享组件，为了避免一刀切的问题，如果用例和需求确实不同（例如，批处理和交互式查询处理），仍可以提供不同的计算引擎。对于这些情况，可以更关注可组合性以及在不同层之间尽可能多地重用组件。例如，交互式和批处理引擎的存储编码或格式没有理由不同。
  • 一致的APIs，用户通常需要与不同的引擎互动以完成工作。如果引擎公开一致的API，可以降低用户的学习曲线，使他们更加高效。同样，一致的API使组件集成更容易，从而促进模块化和可重用性。

这些原则的优势有三个方面：

  • 工程效率，更多的工程师可以在这些（较少数量）系统和组件上工作，防止我们重新发明轮子，将领域特定知识巩固在更少的专业团队中，更加高效，行动更快。
  • 更快的创新， 维护更少的系统可以减少运营负担，并允许工程团队专注于新功能、优化和其他改进，有利于创新。
  • 更好的用户体验，用户现在可以期望在这些系统之间具有一致的语法、语义和功能，减少他们的学习曲线并提高生产力。

Shared Foundations的工作主要集中在几个主要的融合领域，即：计算引擎（交互式，批处理，流处理），SQL方言，存储（存储格式，元数据）、和执行引擎（Execution Engine）。下图说明了层与范围、挑战、期望的最终状态和项目之间的映射。

shared-foundations

Compute Engine Convergence

Compute Engine Convergence包含三个方面的工作，交互式，批处理，流处理引擎的统一。

交互式引擎融合

对于交互式引擎融合引擎应该

  • 提供完整的SQL支持，包括复杂的查询和数据模型
  • 能够直接在湖仓库数据上运行，无需额外的副本
  • 提供低查询延迟，通过将大部分数据存储在内存或本地SSD中来实现
  • 支持实时数据。

交互式引擎融合工作是基于Presto的，因为该系统提供了上述大部分属性，通过亲和性和本地缓存缩小了Presto和其他现有系统之间的性能差距。通过引入智能分层缓存，从而将最常用的数据和元数据保存在工作节点和协调器的本地内存和SSD中，我们实现了大部分常见查询模式的一个数量级的加速。这种加速达到或超过了现有系统在较少硬件上的性能。

Near Real Time Data

存储的另一个重要方面，尤其是对于交互式分析用例，是近实时数据的可用性。虽然实时摄入是大多数数据湖仓库（如BigQuery、Databricks和 Snowflake）的常见功能，但由于某些交互式分析引擎中缺乏离散存储和 Metastore 中必要的功能，Meta数据湖仓库仅限于批量数据。随着Presto支持直接在Hive 数据上进行交互式查询，将实时数据导入湖仓库后便可以进行查询。

FBETL（Meta的摄入系统）已经支持日志数据的连续摄取。然而，查询引擎无法访问这些数据，因为只有在该分区的所有数据可用时，分区才会注册到Metastore 中，这往往是基于小时或每日边界。为了解决这个问题，在Metastore中引入了一个额外的分区状态（“open”），并在数据开始到达时立即注册分区。Presto现在能够在数据降落后立即查询数据，用于临时分析和仪表板，尽管批处理管道继续依赖“closed”分区信号，以确保它们始终在完整数据集上运行。这种设计允许相同的数据集被实时查询和批处理管道消费，避免了繁琐的迁移。最后，我们还加强了摄取提交协议，以确保实时数据的精确一次交付。

接近实时摄取功能已经在湖仓库的许多数据集中启用，从记录到可查询，数据新鲜度在几分钟内。元数据改进：Presto 元数据演变的一个重要方面是关于性能和提高元数据的可缓存性。Hive Metastore已经利用 memcache来减少MySQL访问的延迟，但在更新事件中放宽一致性保证的情况下，客户端无法缓存此元数据。重建了元数据层和API，为每个在更新事件中发生变化的元数据引入了版本概念。Presto利用这个版本信息引入了元数据缓存，大大提高了重复查询的元数据延迟。

注：这里Near Real Time Data的元数据优化感觉是一个残次实现的类似Iceberg，Delta Lake，比如元数据version不就是Iceberg的Snapshot嘛

批处理引擎融合

Meta面临的情况是，有两个引擎（Presto，Spark），两种不同/不兼容的方言，没有一个引擎具备运行所有工作负载所需的功能集，

  • Presto的流式架构在机器故障方面的恢复能力不足，大部分较大、运行时间较长的管道都倾向于使用Spark，因为它具有更好的恢复特性，包括更具可扩展性的shuffle实现。
  • 更进一步复杂化的是，由于Spark和Presto具有不同的资源模型，机器之间没有互换性，必须为Presto和Spark分别提供批处理能力。

Presto on Spark项目通过重构Presto前端（解析器、分析器、优化器、规划器）和后端（表达式计算，算子和I/O）库，并将它们分别嵌入到Spark驱动程序和工作节点中来实现这一点。通过在前端和后端运行完全相同的代码，Presto on Spark保证了与PrestoSQL 100%的兼容性，使用户可以在交互式、即席查询和批处理用例之间无缝切换，无需重写查询。同时，由于它运行在Spark RDD运行时上，并使用Meta的可扩展Cosco shuffle基础设施，Spark的可扩展性和恢复能力得到了充分利用（尤其是细粒度任务重试），使Presto on Spark能够使用与Spark相同的资源池运行大规模和长时间运行的管道。Presto on Spark的一个令人惊讶的附带好处是延迟优势；事实证明，许多管道实际上可以在Presto on Spark上运行得更快，主要原因是Presto on Spark能够为给定查询分配更多的shuffle分区。

注：MPP的Push-based Pipeline模型在容错方面与Spark这种Stage-By-Stage的Pull-based的volcano模型比有天然的劣势，但是我对Presto On Spark的前景不看好。

流处理引擎融合

Meta的流处理也因多种原因逐步演变，导致了一个碎片化的生态系统，主要的两个主题是编程语言（C++ vs. Java vs. PHP）和抽象级别（低级过程式 vs. 高级类SQL声明式API）。传统的数据湖仓库栈由多个流处理引擎组成：Puma（Java，声明式），Stylus（C++，低级别），以及其他具有不同抽象级别（声明式，过程式）和实现语言（C++，Java，PHP）的组合。

这种碎片化导致了多个挑战。对于终端用户，他们需要处理不一致的语言设计和实现细节，例如，一些解决方案是在至少一次语义下运行的，而另一些是在至多一次语义下运行的。对于开发人员，维护多个栈的负担很重，而且像数据管理（eg，lineage, schematization,
and privacy enforcement）这样的新功能需要多次实现。为了摆脱现有的技术债务，并继续支持实时机器学习等新兴需求，Meta构建了下一代流处理平台，称为XStream。

在开发XStream的过程中，一开始的设计是提供一个基于Dataframe的声明式语言，底层使用生成的C++代码进行表达式求值，并将预定义的模板代码粘合以执行诸如连接和窗口聚合等常见转换。虽然Dataframe语言具有很高的表现力，但对于终端用户来说，学习曲线并不平缓，而且SQL支持一直在不断要求。用C++模板编译生成的代码耗时，而且编译错误很难让用户理解。因此，决定与CoreSQL集成，添加流处理扩展，并将SQL作为主要的语言体验。另一方面，用基于Velox解释执行替换了代码生成，这不仅能够利用一个统一的库进行执行，还提供了性能优势。如今，XStream支持从SQL、机器学习、函数即服务和低级系统协调等多种用例。它使我们能够弃用Puma，从而消除另一种SQL方言（PQL）。

SQL Dialect Convergence

Meta有6种SQL方言正在被积极使用：Presto SQL、HiveQL（在 Spark 中）、PQL（Puma）、Scuba SQL、Cubrick SQL和MySQL。这给用户带来了非常陡峭的学习曲线，并且在为各种方言添加功能时浪费了大量的工程师精力，最后融合为两种MySQL和Presto SQL：

  • MySQL是唯一用于OLTP应用程序的方言，在Meta中无处不在，因此放弃它是不切实际的。然而，它在类型系统和可扩展性方面存在局限性，使其不适合分析应用程序。
  • 在分析方言中，Presto的SQL方言是一个更好的选择，因为它已经广泛采用，具有干净的标准兼容设计，并支持复杂类型、丰富功能和可扩展性。我们将其用于所有分析应用程序，扩展用于流式处理、图形和其他用例，并在Meta内部将其命名为CoreSQL。

然而，难点在于如何在不同的引擎之间实现兼容性（谷歌已经通过 ZetaSQL 实现了这一点），从高层次来看，需要两个组件：

  1. 一个SQL解析器和分析器（前端），负责解析和分析查询，以及创建和验证查询计划。对于这个，我们已经有了一个Java实现（Presto）和一个Python 实现（用于开发者工具）。Meta决定将Python实现重写为 C++，以获得更好的性能和与C++引擎的更好集成。目前正在进行的工作是将其进一步简化为一个带有Java 绑定的单一 C++ 库。
  2. 一个函数和操作符库（后端），提供语言的规范实现。同样，已经有了来自 Presto的Java 实现，还开始了一个雄心勃勃的努力，从头开始用C++重写执行引擎作为一个库（Velox），以获得最大的性能和跨引擎的可移植性。
    通过将前端和后端作为库提供，引擎采用CoreSQL作为跨引擎的标准方言变得更容易，新的整合引擎用于交互式分析、批处理（Presto on Spark）和流式处理（XStream）都支持CoreSQL作为唯一的SQL方言。

Storage Convergence

ORC的编解码收敛

Meta传统上使用ORC作为湖仓库的列式格式，内部的ORC变体，名为DWRF，具有更多的功能，如更好地支持大型映射和更细粒度的加密。在Meta内部使用此格式的计算引擎的有机增长导致了编解码器库的碎片化空间。这些编解码器的两个Java实现存在，一个用于Spark和DiGraph，一个用于Presto，还有一个名为DWIO的C++实现，主要用于ML应用程序。尽管这三个库遵循DWRF标准，但它们各自都有局限性。

  • 首先，将Java编解码器合并为一个，使用Presto编解码器作为基础，因为它的性能更高，而且已经开源，然后逐步将任何缺失的功能合并到其中。然后，我们将Spark、DiGraph和其他系统切换到新的编解码器。
  • 其次，将DWIO库重构为Velox，并添加了所有进入Java的功能和优化。这现在作为Velox的一部分在开源中提供。

新的ML优化文件格式。

将分析和ML表一起存储在数据湖仓库中在管理和集成方面是有益的（例如，可以使用标准的分析查询引擎（如Presto和Spark）分析和处理ML表），但它也带来了一些独特的挑战。例如，ML表的增长速度越来越快，比分析表多一个数量级。ML表通常也更宽，往往有数万个特征，通常存储为大型映射。DWRF格式最紧迫的问题是元数据开销；我们的ML用例需要大量的特征（通常存储为巨大的映射），而DWRF映射格式，尽管经过优化，但元数据开销太大。除此之外，DWRF还有其他与编码和条纹结构相关的限制，这些限制在向后兼容的方式下非常难以修复。因此决定构建一个新的列式文件格式，以满足下一代数据堆栈的需求；具体来说，从一开始就针对ML用例，但不牺牲任何分析需求，即Alpha的新格式。

Alpha具有几个显著特点，使其特别适用于混合分析和机器学习训练用例。它具有一种自定义的元数据序列化格式，解码速度明显更快，尤其是对于非常宽的表格和深度映射，此外还采用了更现代的压缩算法。它还提供了更丰富的编码集和一种自适应编码算法，可以根据历史数据模式智能地选择最佳编码，通过编码历史回环数据库。对于许多常见数据类型，Alpha每列需要的流数量较少，使得读取合并更容易，节省I/O，尤其是对于HDD。Alpha是用现代C++从头开始编写的，可以方便地在未来进行扩展。Alpha已经在生产环境中部署，用于几个重要的机器学习训练应用，并在解码方面表现出比ORC高2-3倍的性能，同时具有可比的编码性能和文件大小。

Execution Engine Convergence

Meta数据湖仓库的有机演变为执行引擎创造了一个碎片化的生态系统。这导致了几十个专门的实现，它们之间几乎没有共享，用不同的编程语言编写，由不同的工程团队维护，并且在很大程度上向用户提供不一致的语义。例如，一项非正式的内部调查发现，至少有12个不同的简单字符串操作函数substr()的实现，它们呈现出不同的参数语义（基于0或基于1的索引）、空值处理和异常行为。

为了解决这些挑战，创建了Velox，一种新颖的最先进的C++数据库加速库，提供高性能数据处理组件，旨在统一不同计算引擎之间的执行引擎。在常见的使用场景中，Velox将完全优化的查询计划作为输入，并使用本地主机中可用的资源执行所描述的计算。Velox将以前仅在单个引擎中找到的优化民主化，提供了一个可以实现跨引擎一致语义的框架。这减少了工作重复，促进了可重用性，并提高了整体效率和一致性。Velox正在积极开发中，但它已经在Meta的十几个数据系统中处于不同阶段的集成，包括Presto、Presto on Spark、XStream、FBETL（我们将数据摄取到仓库的系统）、Scribe 以及其他内部 ML 系统，用于特征工程和数据预处理，甚至还有事务系统。

展望

虽然上述部分中的许多工作已经投入生产，但要充分利用基于共享基础设施的新型现代化架构的优势，仍然需要大量工作。一个正在进行的调查领域是为用户自定义函数提供统一支持，因为跨引擎的UDF支持API存在很大的不兼容性，并提供不一致的用户体验。为了扩展 SQL 的语言整合工作，还在探讨其他非SQL API（如数据框接口和其他 DSL，如TorchArrow、Spark Dataset API 和 Pandas）是否可以统一，因为随着ML和数据科学的普及，它们变得越来越普遍。

组件化和可组合性是数据管理的未来。在技术栈的更深层次，已经开始研究查询优化器是否可以整合。虽然现有技术表明优化器与引擎的运行时和物理能力紧密相连，它们至少可以共享一个底层框架。Apache Calcite和Orca等项目为我们提供了先例，但这些组件可以整合的程度仍然是一个悬而未决的问题。最后，随着Velox 成为Meta内外执行的标准，正在探索如何增强Velox以利用硬件加速器，能够在硬件发展的同时调整所有引擎。