利用 dbt 和 AWS Glue 构建与管理现代数据架构

关键要点

在这篇文章中，我们介绍了 dbtglue，一个全新的“可信”dbt适配器，旨在帮助数据工程师利用 dbt 和 AWS Glue 构建现代数据栈。dbtglue 使得用户能够在云数据湖中轻松地进行数据转换，支持多种开放表格式，并且经过 AWS 与 dbt Labs 的战略合作认证为可信适配器。以下是核心内容：

dbtglue 允许将 SQL 代码作为 Spark SQL 在 AWS Glue 交互式会话中运行。该适配器管理 AWS Glue 数据目录中的表定义。支持多种开放表格式，如 Apache Hudi、Delta Lake 和 Apache Iceberg。提供细粒度的访问控制，支持 AWS Lake Formation 权限设置。

dbt 是一款开源 SQL 优先的模板引擎，允许用户以 Python 和 SQL 编写可重用且可扩展的数据转换。它专注于提取、加载、转换ELT或提取、转换、加载ETL过程中的转换层，通过特定的引擎适配器实现提取和加载功能。dbt 使数据工程师、数据科学家和分析工程师能够使用 SQL 选择语句定义业务逻辑，从而消除编写冗长数据操作语言DML和数据定义语言DDL表达式的需求。通过遵循模块化、可移植性、持续集成和持续交付CI/CD及文档化等软件工程最佳实践，dbt 使数据工程师能够快速、高效地协作部署分析代码。

随着数据湖越来越成为大型企业数据项目的解决方案，我们看到客户强烈希望将 dbt 的适用范围扩展至基于云的数据湖。2022 年，AWS 发布了一个名为 dbtglue 的 dbt 适配器，它允许数据工程师在 AWS Glue 交互式会话中，利用 dbt 为云数据湖、数据仓库和数据库提供服务，按需付费。

现在，我们兴奋地宣布 dbtglue 已成为经过认证的适配器，基于我们与 dbt Labs 的战略合作。值得注意的是，经过认证的适配器并非由 dbt Labs 维护，但 dbt Labs 对其在生产环境中使用的推荐感到满意。

主要功能

dbtglue 适配器的关键功能包括：

功能详情执行 SQL作为 Spark SQL 在 AWS Glue 交互式会话中运行管理表定义在 AWS Glue 数据目录中管理支持开放表格式如 Apache Hudi、Delta Lake 和 Apache Iceberg细粒度访问控制支持 AWS Lake Formation 权限

此外，dbtglue 适配器设计的目的是通过多个技术优化资源利用率，充分发挥 AWS Glue 交互式会话的优势。

本文将展示 dbtglue 适配器如何帮助您的工作负载，以及如何利用 dbt 和 AWS Glue 构建现代数据栈。

常见用例

一个常见的用例是在大型企业中，由中心分析团队负责监控运营效率。他们将应用程序日志导入亚马逊简单存储服务Amazon S3数据湖中的原始 Parquet 表。此外，他们还从运营系统中提取有组织的数据，这些数据捕获了公司组织架构和各种运营组件的成本，并将其存储在原始区域中，使用 Iceberg 表以保持原始架构，便于数据的访问。该团队利用 dbtglue 构建一个经过优化的用于商业智能BI的金模型。金模型将技术日志与计费数据结合，并按业务单元组织指标。借助 Iceberg 的数据仓库风格建模功能，金模型满足了数据湖中性能优化的 BI 分析需求。Iceberg 和 dbtglue 的结合，使团队能够高效构建准备好供最终用户使用的数据模型。

另一个常见用例是当公司中的分析团队在 S3 数据湖中创建新数据产品，以增强其现有数据。例如，该公司位于欧洲，该数据产品必须遵循通用数据保护条例GDPR。因此，公司利用 Iceberg 满足诸如被遗忘权和数据删除的需求。由于与 AWS Glue 和 Iceberg 的兼容性，以及 dbtglue 适配器在使用此存储格式时所带来的便利性，这家公司使用 dbt 来对其现有数据湖的数据产品进行建模。

dbt 和 dbtglue 工作机制

以下是 dbt 的关键特性：

ios加速器试用三天项目 dbt 项目在 staging、模型、权限和适配器上施加了顶级结构。项目可以被推送到 GitHub 进行版本控制。SQL dbt 依赖 SQL 选择语句来定义数据转换逻辑。相较于原始 SQL，dbt 提供模板化 SQL使用 Jinja来实现代码的模块化。通过避免在多个地方复制粘贴 SQL，数据工程师可以定义模块化转换并在项目中的其他地方调用它们，促进协作。模型 dbt 模型主要以 SELECT 语句形式编写，并保存为 sql 文件。数据工程师为其数据表示定义 dbt 模型。要详细了解，可以参考 关于 dbt 模型。 物化 物化是将 dbt 模型持久化到仓库中的策略。dbt 中内置了五种物化类型：表、视图、增量、临时和物化视图。要了解更多内容，请参考 物化 和 增量模型。数据血统 dbt 跟踪数据血统，允许您理解数据的来源以及如何在不同转换中流动。dbt 还支持影响分析，这有助于识别更改的下游影响。

数据流的高层次流程如下：

数据工程师将数据从数据源摄取到原始表中，并为原始表定义表定义。数据工程师使用模板化 SQL 编写 dbt 模型。dbt 适配器将 dbt 模型转换为数据仓库兼容的 SQL 语句。数据仓库运行 SQL 语句以创建中间表、最终表、视图或物化视图。

以下图表展示了体系结构。

dbtglue 工作步骤：

dbtglue 适配器将 dbt 模型转换为 Spark SQL 兼容的 SQL 语句。AWS Glue 交互式会话运行 SQL 语句以创建中间表或最终表、视图或物化视图。dbtglue 支持 csv、parquet、hudi、delta 和 iceberg 作为 fileformat。在 dbtglue 适配器上，表或增量通常用于目标的物化。有三种增量物化策略，合并策略要求使用 hudi、delta 或 iceberg。在其他两种策略中，append 和 insertoverwrite 可以使用 csv、parquet、hudi、delta 或 iceberg。

以下图表展示了这一架构。

示例用例

在本文中，我们将使用来自 纽约市出租车记录 数据集的数据。该数据集可在 AWS 开放数据注册中心 中找到，存储了原始的 Parquet 表记录，包含出租车行程记录。

我们的目标是创建以下三个表，基于原始表计算出相应的指标：

silveravgmetrics 基于 2016 年 NYC 出租车开放数据的基本指标goldpassengersmetrics 根据 silver 指标表计算的每位乘客的指标goldcostmetrics 根据 silver 指标表计算的每位行程的成本指标

最终目标是创建两个设计良好的金表，以 Iceberg 格式存储已经聚合的结果，供 Amazon Athena 进行临时查询。

前置条件

此说明需要以下前置条件：

一个具有运行 AWS Glue 交互式会话和 dbtglue 适配器所需的所有权限的 AWS 身份和访问管理IAM角色用于存储与 NYC 出租车记录数据集相关元数据的 AWS Glue 数据库和表一个用于作为输出的 S3 存储桶，以存储处理后数据一个 Athena 配置工作组和 S3 存储桶以存储输出以探索数据集一个 AWS Lambda 函数作为 AWS CloudFormation 自定义资源创建，用于更新 AWS Glue 表中的所有分区

有了这些前提条件，我们模拟了数据工程师已经从数据源摄取数据到原始表，并为原始表定义了表定义的情况。

为了方便使用，我们准备了一个 CloudFormation 模板。该模板自动部署所有所需的基础设施。要创建这些资源，请选择 Launch Stack 在 useast1 区域，并按照说明操作：

安装 dbt、dbt CLI 及 dbt 适配器

dbt CLI 是用于运行 dbt 项目的命令行接口。它是免费的作为 开源项目 提供。使用以下代码安装 dbt 和 dbt CLI：

bash pip3 install nocachedir dbtcore

有关更多信息，请参考 如何安装 dbt、什么是 dbt？ 和 视点。

使用以下代码安装 dbt 适配器：

bash pip3 install nocachedir dbtglue

创建 dbt 项目

完成以下步骤以创建 dbt 项目：

运行 dbt init 命令创建并初始化一个新的空 dbt 项目：

bash dbt init

为项目名称输入 dbtgluedemo。选择数据库，输入 glue。

现在空项目已经创建，目录结构如下：

bash cd dbtgluedemo tree READMEmd analyses dbtprojectyml macros models example myfirstdbtmodelsql myseconddbtmodelsql schemayml seeds snapshots tests

创建源

下一步是创建源表定义。我们添加 models/sourcetablesyml，内容如下：

yamlversion 2

sources name datasource schema nyctaxi

tables   name records

此源定义对应于我们在 CloudFormation 堆栈中创建的 AWS Glue 表 nyctaxirecords。

创建模型

在这一步中，我们创建一个 dbt 模型，表示行程持续时间、乘客人数、行程距离和收费总额的平均值。完成以下步骤：

创建 models/silver/ 目录。创建文件 models/silver/silveravgmetricssql，内容如下：

sql WITH sourceavg AS ( SELECT avg((CAST(dropoffdatetime AS LONG) CAST(pickupdatetime AS LONG))/60) AS avgduration avg(passengercount) AS avgpassengercount avg(tripdistance) AS avgtripdistance avg(totalamount) AS avgtotalamount year month type FROM {{ source(datasource records) }} WHERE year = 2016 AND dropoffdatetime IS NOT NULL GROUP BY year month type ) SELECT FROM sourceavg

创建文件 models/silver/schemayml，内容如下：

yaml version 2

models name silveravgmetrics description This table has basic metrics based on NYC Taxi Open Data for the year 2016

   columns      name avgduration       description The average duration of a NYC Taxi trip      name avgpassengercount       description The average number of passengers per NYC Taxi trip      name avgtripdistance       description The average NYC Taxi trip distance      name avgtotalamount       description The average amount of a NYC Taxi trip      name year       description The year of the NYC Taxi trip      name month       description The month of the NYC Taxi trip      name type       description The type of the NYC Taxi

创建 models/gold/ 目录。创建文件 models/gold/goldcostmetricssql，内容如下：

sql {{ config( materialized=incremental incrementalstrategy=merge uniquekey=[year month type] fileformat=iceberg icebergexpiresnapshots=False tableproperties={formatversion 2} ) }} SELECT (avgtotalamount/avgtripdistance) AS avgcostperdistance (avgtotalamount/avgduration) AS avgcostperminute year month type FROM {{ ref(silveravgmetrics) }}

创建文件 models/gold/goldpassengersmetricssql，内容如下：

sql {{ config( materialized=incremental incrementalstrategy=merge uniquekey=[year month type] fileformat=iceberg icebergexpiresnapshots=False tableproperties={formatversion 2} ) }} SELECT (avgtotalamount/avgpassengercount) AS avgcostperpassenger (avgduration/avgpassengercount) AS avgdurationperpassenger (avgtripdistance/avgpassengercount) AS avgtripdistanceperpassenger year month type FROM {{ ref(silveravgmetrics) }}

创建文件 models/gold/schemayml，内容如下：

yaml version 2

models name goldcostmetrics description This table has metrics per cost based on NYC Taxi Open Data

   columns      name avgcostperdistance       description The average cost per distance of a NYC Taxi trip      name avgcostperminute       description The average cost per minute of a NYC Taxi trip      name year       description The year of the NYC Taxi trip      name month       description The month of the NYC Taxi trip      name type       description The type of the NYC Taxi  name goldpassengersmetrics   description This table has metrics per passenger based on NYC Taxi Open Data   columns      name avgcostperpassenger       description The average cost per passenger for a NYC Taxi trip      name avgdurationperpassenger       description The average duration per passenger for a NYC Taxi trip      name avgtripdistanceperpassenger       description The average NYC Taxi trip distance      name year       description The year of the NYC Taxi trip      name month       description The month of the NYC Taxi trip      name type       description The type of the NYC Taxi

删除 models/example/ 文件夹，因为它只是 dbt init 命令创建的示例。

配置 dbt 项目

dbtprojectyml 是 dbt 项目的重要配置文件，包含以下代码：

yamlmodels dbtgluedemo # 表示适用所有在 models/example/ 下的文件 example materialized view

我们将 dbtprojectyml