Moeve 用 Athena 标准化数据湖中的 dbt 运行

随着组织规模扩大，数据处理往往会在团队、环境和编排工具之间变得碎片化。这种碎片化会导致模式不一致、逻辑重复、成本可见性有限以及运营开销增加。

在 Moeve，我们也并不例外。我们的分析团队使用 dbt 构建转换，这是一款开源工具，它将转换定义为 SQL 模型，解析它们之间的引用，并确定它们的运行顺序。dbt 描述要转换什么，但它并不定义项目在哪里或如何运行。我们把这一决定留给每个团队，随着项目数量增长，我们发现碎片化的流水线、不一致的计算引擎以及有限的成本可见性拖慢了每个项目。将我们的 dbt 运行标准化在 Amazon Athena 上，正是我们摆脱这一困境的方式。

本文介绍了我们在 Athena 上构建的集中式无服务器解决方案的架构，该方案将新 dbt 项目的入门时间从数天缩短到约 15 分钟。该解决方案集中化了 dbt 在我们各数据湖中的运行方式，同时与编排保持松耦合。它使用 Amazon Athena 作为默认处理引擎、一个集中式 dbt 启动器，以及一个共享事件总线用于下游编排。

在接下来的章节中，我们将解释如何使用 AWS Step Functions 和 AWS Fargate 将 dbt 运行与编排解耦，为什么 Athena 从成本和运营角度来看适合我们的工作负载，将运行参数存储在 Amazon DynamoDB 中而不是流水线代码中如何从入门流程中移除了基础设施部署步骤，以及将结果发布到 Amazon EventBridge 如何让我们的运行层和编排层能够独立演进。

大规模运行 dbt 的挑战

在 dbt 启动器之前，我们的 dbt 运行已经朝不同方向发展。运行逻辑嵌入在项目特定的流水线中，编排和处理紧密耦合，团队为可比的工作负载选择了不同的计算引擎，并且我们对运行参数和重试没有一致的治理。

随着 dbt 项目数量增加，这使得我们难以强制执行一致的标准，也难以在不触及每条流水线的情况下演进解决方案。我们需要一种方式来标准化我们在各数据湖中的 dbt 运行，将运行项目与决定运行什么解耦，改善成本控制和可观测性，并交付更快、更安全的持续集成和持续交付（CI/CD）迭代。

为什么选择 Amazon Athena 作为默认 dbt 引擎

为 dbt 选择处理引擎是一项基础性架构决策，因此我们首先做出了这一决定。

无服务器处理

Athena 是完全无服务器的。无需预置、扩展或维护集群。我们的团队使用默认的 Athena 定价运行查询，该定价按查询扫描的数据量收费，并为我们提供弹性而无需容量规划。

由于每个数据湖都位于各自的账户中，每个团队都自行决定 Athena 的付费方式。工作负载增长为持续、高并发使用的团队可以转向 Athena 容量预留，而无需更改启动器、其 dbt 配置文件或其项目配置。我们的团队目前还没有人需要这样做，而该架构使这一选择对每个团队保持独立。

我们的工作负载是可预测的，但不是持续性的。每个 dbt 项目在其计划触发或上游数据到达时运行几分钟，然后保持空闲直到下一次触发。这种形态正是无服务器成为我们合适选择的原因。

为什么 Athena 适合我们的解决方案

对 Moeve 而言，决定标准化使用 dbt 和 Amazon Athena 的驱动力，是我们希望创建一个通用转换解决方案，使其能够在多个团队和 AWS 账户中被采用，同时保持运营轻量化。

我们的数据已经存储在 Amazon S3 中，并注册在 AWS Glue Data Catalog 中，这使 Athena 成为自然的处理层。Athena 使我们无需管理集群、容量或基础设施即可运行转换，这对一个支持多个领域的小型中央团队尤为重要。

我们评估了替代处理引擎，但针对我们的工作负载特征和数据量，Athena 在可扩展性、运营简单性和可维护性之间提供了最佳平衡。适配器成熟度是另一个重要因素。dbt Athena 适配器提供了与测试、CI/CD 工作流以及更广泛 dbt 生态系统的强大集成，降低了维护自定义解决方案的运营风险。

Athena 也与我们的架构天然契合。转换通过跨账户角色代入在拥有数据的 AWS 账户中运行，而编排保持集中化。因此，我们标准化了项目跨团队的运行方式，同时使计算靠近数据。

最后，Athena 中的 Apache Iceberg 支持支撑了本文所述的幂等增量处理模型，因此增量加载和历史重处理遵循相同路径，运营开销最小。

优化的增量处理

我们最大的成本不是读取源数据，而是合并到其中。

我们的事实表是数据湖中的 Apache Iceberg 表，并且我们的大多数模型都是增量的。每次运行都会引入新的或更正的记录，并将它们合并到一个可能保存数年历史的目标中。合并必须定位它将要更新的行，而如果目标上没有谓词，查询读取的表数据会远远多于传入数据所能影响的范围。常见做法是使用诸如最近 30 天这样的静态过滤条件，但它在两个方向上都是错误的：对于普通的每日加载来说范围太宽，而一旦出现针对较旧分区的更正，又范围太窄。

该平台不是使用固定窗口，而是从数据中推导谓词。在合并运行之前，它会读取传入数据集中存在的分区列的不同值，并据此构建目标谓词。对于单个分区，它会应用等值谓词；对于小集合，则应用一个 IN列表，对于更大的集合则是一个有界范围。之后，合并操作只读取传入数据可能影响到的分区。

同样的原则也适用于源端。启动器根据该次运行所给定的参数构建源过滤器：一个显式范围、一个任意 SQL 条件，或者当两者都未提供时，从项目配置中获取的默认窗口。因此，输入被限制为每次运行所需的子集。

有两个结果对我们很重要。由于 Athena 按查询扫描的数据量收费，缩小合并两端的数据范围降低了成本，而无需任何团队手动调优单个模型。更重要的是，每日加载和完整历史重处理变成了同一操作，只是输入不同。每次运行都是幂等的，因此无论运行多少次，相同的输入始终产生相同的结果。这从我们的运行手册中消除了处理与重处理之间的区别，并简化了事件响应。

表设计是使这种剪枝成为可能的原因。分区、列式格式和压缩都有贡献，而 AWS 大数据博客文章 Amazon Athena 十大性能调优技巧 涵盖了通用技术。我们特意没有在此发布前后对比数字，因为节省量在很大程度上取决于分区设计和数据分布，在没有大量上下文的情况下，单个数字会产生误导。

架构概述

Moeve 构建了一个集中式 dbt 启动器，以统一方式运行 dbt 作业，无论项目或目标数据湖是什么。

架构图

图 1：集中式 dbt 启动器和跨账户运行流程

在高层次上，该架构包括：

• AWS Step Functions 用于控制运行生命周期。
• AWS Fargate 用于在隔离的临时容器中运行 dbt。
• Amazon Athena 作为默认的 dbt 处理引擎。
• Amazon DynamoDB 用于存储 dbt 项目配置。
• Amazon EventBridge 用于发布运行结果。

每次 dbt 运行都遵循相同的契约，这为我们提供了一致性，并减少了我们必须维护的运营面。

跨账户处理模型

该解决方案在集中式账户中运行，同时在特定领域的数据湖账户中执行转换：企业、营销和制造。Fargate 容器承担目标账户中的 dbt-child IAM 角色，因此容器在数据所在之处处理数据，而治理保持集中。

每个数据湖账户保留对其自身 IAM 权限的控制，并管理自己的存储和目录，而不会影响该解决方案。这也就把成本放在了正确的账户中。我们本可以使用 Athena 工作组来归属 Athena 支出，但 Athena 只是查询成本的一部分。Amazon Simple Storage Service（Amazon S3）它所发起的请求以及它触发的 AWS Key Management Service（AWS KMS） 操作也是真实成本，而在拥有数据的账户中运行会把所有这些成本按项目和按运行归属到拥有数据的团队。

使用 dbt 启动器集中化 dbt 运行

为了阻止每个团队发明自己的 dbt 运行方式，我们构建了一个所有团队都使用的单一启动器。dbt 逻辑不再存在于多个管道中，而是每次运行都通过一条定义良好的路径触发。

运行生命周期

启动器是一个 AWS Step Functions 状态机。它接收带有参数的运行请求，从我们的 dbt 容器镜像启动一个 AWS Fargate 任务，然后该容器承担目标数据湖账户的 IAM 角色。随后 dbt 在 Athena 中运行其 SQL 转换，读取源表并物化目标表。在运行的同时，Elementary，一个开源 dbt 包，记录模型级结果和数据质量测试结果。当运行完成时，启动器会向 Amazon EventBridge 发布完成事件。

状态机可以根据场景以不同方式启动。一些项目按计划运行，另一些在上游数据到达时触发，团队还可以按需请求运行。这些决策由我们的编排层做出，该层向启动器提交标准化的运行请求。因此，无论运行是如何发起的，启动器都专注于运行 dbt 项目。

这个序列对于所有项目和環境都是相同的。启动器只负责运行它被要求运行的项目。它不决定接下来应该运行什么，这一职责被有意通过 Amazon EventBridge 委托给下游消费者。

使用 Amazon DynamoDB 进行配置驱动的运行

我们必须决定项目的运行参数存放在哪里。在管道定义中，更改超时将是一次代码更改、一次审查、一次构建和一次部署，而有时我们需要在事件处理期间更改这个值。我们把参数放在 DynamoDB 中，以项目为键，启动器在每次运行开始时读取它们。这让我们能够将代码部署与运行行为解耦，无需重新部署服务即可更新参数，并在所有 dbt 项目中强制执行一致的默认值。

这些参数本身很有限。它们涵盖目标环境和 AWS Identity and Access Management（IAM）角色、默认处理引擎、允许运行多长时间、失败时重试多少次，以及默认处理多少天的数据。这些值因运营原因而非逻辑原因而变化，这就是为什么我们不想让它们与发布周期耦合。

入职配置的影响比我们预期的要大。现在部署一个新的 dbt 项目只需合并 dbt 模型和一条配置项。无需更改 Terraform，无需基础设施审查，也无需预置任何资源，因为项目所需的计算资源已经存在并且是共享的。过去的多步骤基础设施流水线现在变成了单个 CI 工作流，它在本地验证项目的 SQL 和血缘关系，然后合并并注册项目。我们使用 DuckDB 运行本地验证，它能在两分钟内返回反馈，且不消耗云资源。

治理并没有因此被削弱。谁可以更改配置项，其管控方式与任何其他生产变更相同。变化在于，该变更不再需要通过基础设施部署才能生效。

CI/CD 流水线图

图 2：用于入职和更新 dbt 项目的 CI 工作流

Athena 仍然是权威引擎。本地验证可以捕获 Jinja 错误、未解析的引用和明显的 SQL 错误，但 Athena 特有的行为、跨账户权限以及与 AWS Glue Data Catalog 的交互只有在目标环境中才能得到验证。与团队明确这一边界很重要，这样 CI 通过就不会被解读为一种保证。

使用 Amazon EventBridge 发布运行结果

dbt 运行完成后，启动器向一个中央 Amazon EventBridge 总线发布结构化事件，记录运行是否成功、覆盖了哪个项目和数据源、何时开始和结束、耗时多久以及更新了哪些数据集。启动器并不知道哪些消费者订阅了这些事件。

这种事件驱动的方法让我们在运行项目和编排后续操作之间实现了松耦合，同一信号可以拥有多个下游消费者，并且两层都可以独立演进。

在 Moeve，主要消费者是我们的编排层，它将数据集之间的依赖关系建模为图。每个完成事件都告诉它某个节点现在已经是最新的，因此它可以确定哪些下游项目的所有输入都已就绪并启动它们。该消费者没有特殊地位。一个 AWS Lambda 函数、监控仪表板或通知集成都可以订阅相同的事件，而无需对启动器做任何更改。

验证和可观测性

因为每个项目都遵循相同的生命周期，我们在同一个地方而不是按流水线获得验证和可观测性。Step Functions 显示任何运行的状态以及它在哪一步失败，错误处理和重试只需定义一次。Fargate 日志包含容器运行时细节。dbt 和 Elementary 报告模型级结果和数据质量测试结果。总线上的完成事件是项目已完成及其产出的可审计记录。这些层共同为我们提供了运维可见性，而无需将各组件相互耦合。

成本控制和资源清理

该平台端到端都是无服务器的，这使空闲成本接近于零，并消除了一类运维错误。Fargate 任务在运行时创建，结束时销毁，因此无需维护长时间运行的容器。Athena 没有持久计算资源。空闲的 Step Functions 状态机不产生任何费用。没有任何东西会在无意中保持运行，这在平台上的项目数量持续增长时尤为重要。

结果

下表中的指标是我们的团队日常会注意到的，也是该解决方案能够由一个小型中央团队维护的原因。

结论

标准化我们运行 dbt 的方式，结果发现其依赖的不是 dbt 本身，而是清晰地定义两个边界。

第一个是启动器的契约：参数传入，事件传出。我们在构建内部实现之前就定义了这个接口，并且在实现多次更改的情况下它一直保持稳定。第二个是运行项目与决定接下来运行什么之间的分离。让启动器在不知道消费者的情况下发布事件，正是我们的编排层可以被重建而启动器保持原样的原因，而且启动器从未为了适应新的编排需求而被修改。

两个较小的决策承载了比我们预期更大的分量。将运行参数保存在 DynamoDB 而不是流水线定义中，意味着超时、重试和引擎选择可以在不部署的情况下更改，这在事件响应期间很有价值。让每次运行都幂等，消除了处理和重新处理之间的区别，因此每日加载和完整历史重处理是相同操作，只是输入不同。

Athena 是无服务器的，因此我们的团队不需要自己搭建基础设施。这正是让每个人都能以标准化方式运行 dbt 变得切实可行的原因，也是新项目入职从数天缩短到约 15 分钟的原因。

如果你的组织在多个账户和团队中运行 dbt，请将 Amazon Athena 与运行项目的组件的清晰契约搭配使用：固定参数传入，发布事件传出。

资源

• Amazon Athena 文档
• AWS Step Functions 开发人员指南
• Amazon EventBridge 用户指南
• Amazon Athena 的 10 大性能调优技巧
• 在 Amazon Athena 中使用 Apache Iceberg 表
• dbt 文档