基于 S3 的 FaaS 数据处理管道

作者制作的这张图片。

简介

AWS Lambda 是一个有趣的服务。

我在2021年第一次使用它，使用起来非常顺畅。我写了一些代码并配置了触发条件后，就这样。

每当有新的文件到达S3时，我的Lambda函数就会激活，执行逻辑，然后再次休眠。我完全不用担心任何基础设施的问题，

这让我好奇：我能否通过我的数据流也达到同样的简洁。

如果不需要搭建Airflow环境或配置Spark集群会怎么样？如果我可以像定义一个AWS Lambda函数一样定义工作流逻辑，并通过某种方式，输入数据会变成期望的输出结果吗？岂不更好？

这周，我们要深入了解Bauplan，一个能实现这个愿望的解决方案。

概要:

函数即服务（FaaS）是一种云计算模式，允许开发人员根据事件运行代码，而无需自己管理基础架构。它支持无服务器架构，其中云提供商处理资源调配、突发负载的扩展以及执行任务，使工程师能够专注于编写逻辑。著名的例子是 AWS Lambda 等。

作者的图片。

FaaS 让开发人员的工作更容易。

Bauplan（https://www.bauplanlabs.com/），纽约和旧金山的团队认为，FaaS模型也能让数据工程师、分析师、数据科学家，或任何希望与数据打交道的人的工作变得更简单。

作者自制的图片。

仍然，市场上的解决方案无法应对数据工作负载。我们通常将数据流定义为DAG（有向无环图），其中每个节点都是一个函数，从前面的节点接收数据，应用逻辑并输出给后续节点。

作者自制的图片。

将业务逻辑模块化为节点使得开发、协作和测试更加便捷。但在利用现有的FaaS解决方案来实现DAG数据管道时，会遇到一些挑战：

• 可扩展性: 现有的FaaS运行时设计用于简单的、独立的功能，这些功能通常产生较小的输出，例如 webhook。当应用于数据管道时，它们表现出局限性。此外，这些FaaS平台通常会复用实例来处理后续触发。这意味着同一个功能实例可以处理从10GB到1TB的输入，而使用相同的函数。内存不足的错误很常见。

作者绘制的图片。

• 大型中间 I/O：为了实现 DAG（有向无环图）的概念，用户必须将函数串联起来。每个函数充当一个“节点”，接收来自前一个函数的输入并为后续函数生成输出。数据函数通常具有较大的输入和输出，这增加了序列化和反序列化以及在函数之间移动数据负载的成本。流行的 FaaS 平台的串联最佳实践有一定的局限性，因为中间数据帧只能通过对象存储来传递。

这张图片是作者画的。

• 缓慢的反馈回路：数据科学项目具有探索性，并需要快速迭代以验证假设。当前的FaaS平台由于构建时间长和缺乏交互式日志而无法提供所需的交互功能。AWS Lambda仅通过Cloudwatch提供可观测性功能。

这张作者自己创作的图片。

那么说来，Bauplan又是怎么打算解决这些问题的？

Bauplan 函数即服务

Bauplan 是一种为数据管道设计的函数即服务。与其它服务不同的是，Bauplan 为每个运行独立启动和扩展实例。它还承诺提高中间数据交换的效率，并允许用户互动地修改和运行有向无环图。

在设计原则方面：

如图所示，作者的图片。

Bauplan 的目标是让执行过程无状态化，其实例仅在运行期间存在。每次启动时都使用新实例，这使得 Bauplan 能够适应不同的资源需求；同一个管道可以首先在10GB的数据集上运行，之后可以扩展到100GB的数据集。

• 对于基础设施，Bauplan 管道将在云虚拟机（VM）上运行，云虚拟机提供了最高级别的定制化。使用云虚拟机也使 Bauplan 能够提供多种部署模型，例如 BYOC（自带基础设施），客户可以控制数据的存储和处理位置。
• 与其他工具相比，Bauplan 既有数据意识也有运行时意识（即，像 AWS Lambda 这样的无服务器运行时不了解数据的上下文，而编排工具不了解运行时的上下文）。稍后我们将通过运行一些代码来更深入地探讨这种设计。
• Bauplan 为开发人员带来了交互式体验；虽然管道在云端运行，用户仍可以在本地笔记本电脑上进行开发。Bauplan 提供了一个命令行工具（CLI）和一个 Python SDK，以方便用户与系统进行交互。
• 用户通过指定输入表和输出表来定义 Bauplan 中的函数。

建筑

Bauplan（架构计划）有一个控制面（CP）和一个数据面（DP）

作者绘制的这张图片。

• CP提供了多租户API。它仅处理元数据。CP位于Bauplan的VPC内。
• 每个客户有一个DP，即一个或多个可以部署在客户VPC内的云VM。在每个VM上安装了一个Golang二进制文件以启动工作者。这些工作者是唯一能够访问客户数据的Bauplan组件。

为了让开发者获得互动体验，客户和工作者之间有一个双向的gRPC连接。用户可以通过写一些print或logging语句来了解管道运行时发生了什么；虽然代码是在云虚拟机中运行的，但因为这个双向连接，用户可以立即看到结果。

计划（规划）

那么，CP（内容提供商）需要处理元数据信息，但是它的责任是什么呢？

Bauplan 就像一个数据库一样工作；在开始运行管道时，它会将 Python 和 SQL 代码转换成执行计划。当用户请求运行时，代码会被路由到控制平面（CP）。控制平面（CP）会解析代码，并根据函数重构 DAG 拓扑，从而生成一个逻辑计划。

此计划仅表示步骤间的依赖关系及用户指定的所需软件包。重要的是，Bauplan将拒绝运行引用不存在的表的DAG、指向错误快照的DAG或带有格式无效Python代码的DAG。

作者原创的图片。

为了给工人提供运行指示，CP 从逻辑计划形成物理计划。这个物理计划包含了容器化转换函数的运行指令以及将数据帧映射到存储在对象存储中的物理表的指令。在有了物理计划之后，CP 将其发送给工人以开始执行任务。

Bauplan 中的数据存储在对象存储中的 Iceberg 表里，我们很快就会开始研究存储层。

缓存是什么？

如前所述，函数实例仅在执行时存在。数据管道的两次运行将有不同的实例集，去掉“将”字使得句子更加简洁流畅。为了减少重新运行管道时的延迟问题，Bauplan开发了一套稳健的包缓存机制，以避免在不同运行之间重复安装包，从而避免了对PyPI的调用所导致的额外开销。

对于缓存数据，Bauplan的数据感知使得像数据库那样的优化成为可能。

• 重用中间数据框：函数生成中间数据框，Bauplan 跟踪代码和数据的变化情况，以便缓存并重用中间数据框。

作者这张图。

• 仅检索缺失的列 X：第一次运行读取表中的四列，第二次运行则需要这四列加一列 X。Bauplan 会重用缓存中的四列，并仅从数据源下载列 X。

作者绘制的图像。

• 缓存更新： 由于物理数据存储在这些不可变文件中（通过Iceberg元数据），数据框的变更通过数据提交来确定，这样缓存就知道何时需要更新。

作者自己画的图。

数据交换

为了增强功能间的数据交换过程，Bauplan 将中间数据框表示为 Arrow 格式的表。根据官方文档：

_> Arrow列格式包括一种与语言无关的内存数据结构规范、元数据序列化以及用于序列化和通用数据传输的协议。

与 Parquet 或 CSV 这样的格式不一样，它们定义了数据在磁盘上的组织方式，而 Arrow 更关注数据在内存里的组织方式。

作者画的图。

箭形结构在内存中为每一列连续存储值。这种设计对数据分析任务尤其有利，因为这些任务在处理大规模数据集时，通常仅关注数据集中的特定列。

作者自制的图片。

在 Arrow 出现之前，每个系统都使用其内部内存格式。当两个系统进行通信时，每个系统在传输前将其数据转换为标准格式，这会带来序列化和反序列化过程中的转换成本。Apache Arrow 旨在提供一种在单个系统内部处理时非常高效的格式。随着越来越多的系统采用它，它们可以在非常低的成本下共享数据，从而甚至通过共享内存实现零成本共享。

这张图片是作者自己做的。

当 Bauplan 运行管道时，它将选择共享机制：内存或本地磁盘（在同一工作者中的函数）或 Arrow Flight（跨不同工作者）。由于其他方案仅支持 S3 基础的数据交换，利用 Arrow 技术，Bauplan 中在函数之间移动数据的速度可以快上百倍。

通过 Arrow，函数可以从共享内存、内存映射文件或从 gRPC（带 Flight）流式传输数据，这使得函数在处理多种数据源时更具灵活性，无论这些数据源使用何种数据传输机制。

内存映射文件技术是一种将一段虚拟内存空间与文件或类似文件的某个部分建立逐字节映射的技术。这种技术的好处是提升输入输出效率，尤其是在处理大文件时更为明显。

另外，如果下游的任务与上游的任务在同一工作程序上执行，它可以读取Arrow中间数据而无需复制数据。如果有10GB的中间数据且有四个需要读取它的函数，只需要10GB的RAM，而不是10GB乘以4等于的40GB。

储存

Bauplan 不仅止于此；除了 FaaS 数据管道之外，他们还希望通过提供 Iceberg 和 Project Nessie 存储层，提供一个完整的湖仓解决方案。

如果你有一些 Parquet 文件存放在对象存储里，Bauplan 可以用一行代码帮你把这些文件转换成 Iceberg 表格。数据仍然保留在你的 VPC 中；不需要移动到其他地方。

Netflix 开发了 Apache Iceberg 以实现更好的表正确性以及更快的查询规划（比 Hive 更快）。一个 Apache Iceberg 表有三个分层结构。

作者制作的图片。

• 数据层不仅存储表的当前数据，还包括已被删除的文件。
• manifest文件记录数据层内的数据文件。
• manifest列表记录Iceberg表在特定时刻的状态快照。
• 元数据文件记录有关Iceberg表在某一时间点的信息，比如表的结构或最新快照。
• 目录是所有Iceberg数据操作的起点，它为引擎提供元数据指针的位置，指引操作的开始。

和其他表格格式一样，Iceberg 的目标是把数据仓库的能力带到数据湖；其中重要的一点是 ACID 属性。

冰山仅在表级别上确保原子事务。为了将软件开发体验带入数据湖，Bauplan 使用 Project Nessie 来管理 Iceberg 表目录，这是一个开源的版本化元数据目录，允许 Iceberg 进行跨表的事务处理。用户可以同时更新多个表，并确保所有更改都以原子方式完成，实现跨表的全有或全无提交。

作者绘制的

Bauplan让用户能够更轻松地在Nessie和Iceberg表之间无缝切换，通过提供CLI命令和Python SDK工具。用户会觉得就像自己在操作Git仓库一样。

接下来，我们将探索上面提到的所有超赞的Bauplan特性（这里指代具体功能或设计），并运行一些代码。

运行代码

我们将运行一些Python代码和CLI命令；我已经准备了一个Git仓库，你可以跟着一起做。请确保你将仓库克隆到本地，并进入bauplan_example文件夹里。

首先，我们需要根据仓库里的requirements.txt文件设置一个Python虚拟环境。然后，我们安装bauplan，streamlit和duckdb这三个包。

接下来，我们需要Bauplan API密钥，该密钥可以让你访问Bauplan沙盒环境。你可以通过这里联系Bauplan获取密钥。

设置你的bauplan密钥值为 “your_bauplan_key”

接着，我们将运行一些 bash 脚本来设置环境；让我们把这些脚本设置为可执行。

chmod -R +x scripts/

将scripts/目录下的所有文件设置为可执行权限

Bauplan专为仅在云中运行而设计，以确保数据开发周期的完全可审计性和安全性。他们要求我们将数据存储在对象存储中，以便能够支持导入到Iceberg表。

目前，Bauplan 只支持 S3 作为数据源，并且只支持 Parquet 和 CSV 作为文件格式。我们将运行一个脚本，在 S3 存储桶中创建一个桶，并上传一些 CSV 文件，切换到一个分支上，创建一个命名空间（namespace），然后将这些文件导入到 Bauplan Sandbox 中的 Iceberg 表。但在开始前，请确保你已配置好 AWS CLI：

    aws configure # 设置 AWS 访问密钥、密钥和默认区域名称

首先，运行setup.sh，并使用你想要创建的S3桶名称和Bauplan分支。分支名称必须遵循 <your-user-name>.<something> 的格式。

    ./scripts/setup.sh <桶名称> <bauplan 分支名>

这张图片是作者自己做的。

请稍等一下，过一会儿，你就会在目录里看到五个表了。脚本将创建一个名为 adventure 的命名空间。在 Bauplan 中，命名空间就像是一个逻辑容器，帮助组织目录中的表。从现在起，我们将在你的输入分支（input branch）和 adventure 命名空间上进行工作。咱们来看看有哪些输入表：

    bauplan table --namespace adventure

在我们继续之前，让我们先来了解一下输入的数据。我们将要使用来自AdventureWorks示例数据集的五个输入表格：产品表、产品类别表、产品子类别表、销售表和地区表。

AdventureWorks 数据库支持一个虚构的自行车制造公司的标准在线交易处理场景。

这些表格之间的关系是：

作者的这张图片。

在这个项目中，我们将编写一个Bauplan流水线，将这些输入表转换成一个包含fact_sale、dim_product和dim_country的维度数据模型。

作者的图片。

我们利用Bauplan模型来实现数据转换的管道。模型函数以表格数据作为输入，并生成新的表格数据。我们将编写一个DAG，利用duckdb引擎来转换数据：

作者自制的图片。

我们在models.py文件里定义这些模型。非常重要的一点是，Bauplan模型同时了解数据以及运行时的情况。

为了实现运行时感知，你可以使用Bauplan的装饰器来指定每个模型的运行时，例如Python版本及所需的包，如何生成输出，以及支持明确的列选择和过滤下推。

对于数据理解，每个模型都必须有输入，这些输入可以是目录中的表或其他模型。我们就像在使用Python函数中的参数一样来指定并使用这些输入。

这里是 dim_product 模型的代码。正如你所见，Bauplan 知道需要用 Python 3.11 和 duckdb 1.0.0 来运行此模型，该模型的输入数据有 product、product_category 和 product_subcategory。

你可以在这里查看所有模型的代码链接。

这是我从```(https://github.com/vutrinh274/bauplan_example)取到的截图，。

在有了这些模型后，我们就会运行这个流程。

    bauplan run --project-dir pipeline --namespace adventure

注释：此命令用于运行名为adventure的命名空间下的pipeline项目目录。

我们将代码提交给Bauplan，它会进行规划并执行。如果有错误，因为我们的终端与Bauplan之间有双向gRPC连接，错误会实时显示在终端上。

要运行这个管道，我们需要一个bauplan_project.yml文件，其中包含项目的唯一ID以及名称。我在管道文件夹里找到了bauplan_project.yml和models.py这两个文件。

管道完成后，你可以通过再次查看 adventure 命名空间中的输出表格来检查它们。

    bauplan table --namespace adventure (这里的命令是用于特定上下文的，比如在编程或软件命令中使用)

最后，我还让项目更有趣，创建了一个小型Streamlit应用程序，包含了一个顶级的SQL编辑器，用来查询输出表的数据。

    streamlit run streamlit/app.py

运行 Streamlit 应用程序的命令

这里有一个快速演示视频，展示我世界级的SQL编辑软件。

要清理，你可以运行clean_up.sh脚本，自动清理S3桶和Bauplan表。

    ./scripts/clean_up.sh <bucket 名> <分支>

我的想法。

在这个数据就是新黄金的世界里，每家公司都希望拥有处理数据的能力，来推动业务决策。然而，并不是每一家公司都有专门的数据团队。在很多情况下，你可能就是团队里唯一的数据专家。

本文一开始，我有个愿望——它实现了。Bauplan 可以处理通常需要整个基础设施团队来完成的任务。它的目标是提供一种无缝且友好的开发人员使用方式，可以直接在 Python 中处理大规模数据，消除基础设施瓶颈。

当我在使用Bauplan运行一些代码时，我真正被它无缝地从对象存储导入数据文件到Iceberg表中的能力所震撼。设置Iceberg目录（catalog）、配置Iceberg写入器（writer）以及管理物理存储布局通常是一个繁琐的过程，但Bauplan大大简化了这些步骤。

定义数据转换也是一件很愉快的事情，这得益于Bauplan的“模型”理念。我只需修改装饰器中的几行代码，就可以用Python 3.9或3.10运行转换过程。模型的数据感知性使编写转换逻辑变得非常直观——我可以像定义Python函数输入一样轻松地写出转换逻辑。

Bauplan真的非常创新。真的值得一试，特别是如果你是数据工程师、分析师，或者科学家，或者你就是个数据爱好者。

就我个人希望而言，我希望Bauplan能够扩展以支持像Spark或Trino这样的数据处理引擎。无服务器的Spark或Trino集群将是一个巨大的改变。此外，一个强大的SQL编辑器用于查询目录中的数据也会是非常有用的。

结束

感谢您读到这么远。

在本文中，我们将探讨实现数据管道所面临的挑战，使用现有的 FaaS 解决方案，Bauplan 如何解决这些问题，Bauplan 的设计目标和架构，Bauplan 如何提供一个完整的零基础设施湖仓解决方案，其中包括 Iceberg 和 Project Nessie 存储层，最后我们将编写一些代码来构建 Bauplan 管道。

现在说再见吧，下次文章见。

参考

1. Bauplan 文档
2. Jacopo Tagliabue, Tyler Caraza-Harter, Ciro Greco,Bauplan：零拷贝、可扩展的FaaS数据流（Bauplan） (2024)