dbt 允许数据分析师和工程师使用与软件工程师构建应用程序相同的实践来转换他们的数据。dbt是ELT中的T。组织、清洗、去规范化、过滤、重命名和预聚合您的仓库中的原始数据，以便它可以用于分析。

dbt-glue

dbt-glue 包实现了针对AWS Glue的Spark引擎的 dbt适配器 协议。它支持通过新的Glue交互式会话API在Spark上运行dbt。

安装

可以使用以下命令从PyPI安装此软件包

$ pip install dbt-glue

有关更详细（且更有可能是最新的）的信息，请参阅 README

连接方法

配置您的AWS配置文件以用于Glue交互式会话

交互式会话使用了两个IAM主体。

• 客户端主体：从本地客户端调用AWS API（Glue、Lake Formation、交互式会话）的主体（用户或角色）。这是在AWS CLI中配置的主体，可能是相同的。
• 服务角色：AWS Glue用于执行您的会话的IAM角色。这与AWS Glue ETL相同。

阅读 此文档 以配置这些主体。

要享受 dbt-glue 适配器的所有功能，您需要将以下3个AWS管理策略附加到服务角色

服务	所需的管理策略
Amazon S3	AmazonS3FullAccess
AWS Glue	AWSGlueConsoleFullAccess
AWS Lake formation	AWSLakeFormationDataAdmin

本地环境的配置

由于 dbt 和 dbt-glue 适配器与Python版本3.7、3.8和3.9兼容，请检查Python版本

$ python3 --version

配置 Python 虚拟环境以隔离包版本和代码依赖

$ sudo yum install git
$ python3 -m pip install --upgrade pip
$ python3 -m venv dbt_venv
$ source dbt_venv/bin/activate
$ python3 -m pip install --upgrade pip

配置 AWS CLI 的最新版本

$ curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
$ unzip awscliv2.zip
$ sudo ./aws/install

配置 aws-glue-session 包

$ sudo yum install gcc krb5-devel.x86_64 python3-devel.x86_64 -y
$ pip3 install —upgrade boto3
$ pip3 install —upgrade aws-glue-sessions

示例配置

type: glue
query-comment: This is a glue dbt example
role_arn: arn:aws:iam::1234567890:role/GlueInteractiveSessionRole
region: us-east-1
workers: 2
worker_type: G.1X
idle_timeout: 10
schema: "dbt_demo"
database: "dbt_demo"
session_provisioning_timeout_in_seconds: 120
location: "s3://dbt_demo_bucket/dbt_demo_data"

下表描述了所有选项。

选项	描述	必填
project_name	dbt 项目名称。这必须与在 dbt 项目中配置的名称相同。	是
type	要使用的驱动程序。	是
query-comment	一个字符串，用于在 dbt 运行的每个查询中注入注释。	否
role_arn	作为 CloudFormation 模板的一部分创建的交互式会话角色的 ARN。	是
region	您在其中运行数据管道的 AWS 区域。	是
workers	当作业运行时分配给定义的 workerType 的工作进程数。	是
worker_type	当作业运行时分配的预定义工作进程的类型。接受 Standard、G.1X 或 G.2X 的值。	是
schema	用于在 Amazon S3 中组织存储的数据的模式。	是
database	在 Lake Formation 中的数据库。该数据库在数据目录中存储元数据表。	是
session_provisioning_timeout_in_seconds	AWS Glue 交互式会话预配的超时时间（以秒为单位）。	是
location	您的目标数据的 Amazon S3 位置。	是
idle_timeout	AWS Glue 会话空闲超时时间（以分钟为单位）。（会话在空闲指定时间后会停止。）	否
glue_version	此会话要使用的 AWS Glue 版本。目前，唯一有效的选项是 2.0 和 3.0。默认值为 2.0。	否
security_configuration	与此会话一起使用的安全配置。	否
connections	要在会话中使用逗号分隔的连接列表。	否

配置

配置表

当将模型作为 table 物化时，您还可以包括一些特定的 dbt-spark 插件配置选项，以及标准的 模型配置。

选项	描述	必需？	示例
file_format	创建表时要使用的文件格式（parquet、csv、json、text、jdbc 或 orc）。	可选	parquet
partition_by	按指定的列对创建的表进行分区。为每个分区创建一个目录。	可选	date_day
clustered_by	将创建的表的每个分区根据指定的列分割成固定数量的桶。	可选	country_code
buckets	在聚类时创建的桶数	如果指定了 clustered_by，则必需	8

增量模型

dbt 通过其内置配置和物化来提供有用地直观的建模抽象。

因此，dbt-glue 插件在很大程度上依赖于 incremental_strategy 配置。此配置告诉增量物化如何在首次运行之后的运行中构建模型。它可以设置为以下三个值之一

• append（默认值）：插入新记录，而不会更新或覆盖任何现有数据。
• insert_overwrite：如果指定了 partition_by，则用新数据覆盖表中的分区。如果没有指定 partition_by，则用新数据覆盖整个表。
• merge（仅 Apache Hudi）：根据 unique_key 匹配记录；更新旧记录，插入新记录。（如果没有指定 unique_key，则所有新数据都会插入，类似于 append。）

这些策略各有优缺点，我们将在下面讨论。与任何模型配置一样，incremental_strategy 可以在 dbt_project.yml 或模型文件的 config() 块中指定。

注意：默认策略是 insert_overwrite

《append》策略

遵循《append》策略，dbt 将执行一个包含所有新数据的 insert into 语句。这种策略的优点是简单直接，并且适用于所有平台、文件类型、连接方法和 Apache Spark 版本。然而，这种策略 无法 更新、覆盖或删除现有数据，因此很可能会插入许多数据源的重复记录。

源代码

{{ config(
    materialized='incremental',
    incremental_strategy='append',
) }}

--  All rows returned by this query will be appended to the existing table

select * from {{ ref('events') }}
{% if is_incremental() %}
  where event_ts > (select max(event_ts) from {{ this }})
{% endif %}

运行代码

create temporary view spark_incremental__dbt_tmp as

    select * from analytics.events

    where event_ts >= (select max(event_ts) from {{ this }})

;

insert into table analytics.spark_incremental
    select `date_day`, `users` from spark_incremental__dbt_tmp

《insert_overwrite》策略

当在模型配置中指定 partition_by 子句时，此策略效果最佳。dbt 将运行一个 原子 insert overwrite 语句，动态地替换查询中包含的所有分区。在使用此增量策略时，请务必重新选择 所有 相关数据以用于分区。

如果没有指定 partition_by，则 insert_overwrite 策略将原子地替换表中的所有内容，仅用新记录覆盖所有现有数据。但是，表的列模式保持不变。在某些有限情况下，这可能是有益的，因为它在覆盖表内容时最小化了停机时间。此操作类似于在其他数据库上运行 truncate + insert。对于 Delta 格式化表的原子替换，请使用 table 物化（它运行 create or replace）。

源代码

{{ config(
    materialized='incremental',
    partition_by=['date_day'],
    file_format='parquet'
) }}

/*
  Every partition returned by this query will be overwritten
  when this model runs
*/

with new_events as (

    select * from {{ ref('events') }}

    {% if is_incremental() %}
    where date_day >= date_add(current_date, -1)
    {% endif %}

)

select
    date_day,
    count(*) as users

from events
group by 1

运行代码

create temporary view spark_incremental__dbt_tmp as

    with new_events as (

        select * from analytics.events


        where date_day >= date_add(current_date, -1)


    )

    select
        date_day,
        count(*) as users

    from events
    group by 1

;

insert overwrite table analytics.spark_incremental
    partition (date_day)
    select `date_day`, `users` from spark_incremental__dbt_tmp

指定 insert_overwrite 作为增量策略是可选的，因为它是未指定任何策略时的默认策略。

《merge》策略

使用说明：《merge》增量策略需要

• 文件格式：hudi
• AWS Glue 运行时 2，以及作为额外 jar 的 hudi 库

您可以通过在 profiles.yml 中的 extra_jars 选项中添加 hudi 库将 hudi 库作为额外 jar 添加到类路径中。以下是一个示例

extra_jars: "s3://dbt-glue-hudi/Dependencies/hudi-spark.jar,s3://dbt-glue-hudi/Dependencies/spark-avro_2.11-2.4.4.jar"

dbt 将运行一个 原子 merge 语句，其看起来几乎与 Snowflake 和 BigQuery 上的默认合并行为相同。如果指定了 unique_key（推荐），则 dbt 将使用与键列匹配的新记录值更新旧记录。如果没有指定 unique_key，则 dbt 将放弃匹配标准，并简单地插入所有新记录（类似于 append 策略）。

源代码

{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key='user_id',
    file_format='hudi'
) }}

with new_events as (

    select * from {{ ref('events') }}

    {% if is_incremental() %}
    where date_day >= date_add(current_date, -1)
    {% endif %}

)

select
    user_id,
    max(date_day) as last_seen

from events
group by 1

持久化模型描述

自 dbt v0.17.0 起支持关系级别的文档持久化。有关配置文档持久化的更多信息，请参阅 文档。

当配置了适当的 persist_docs 选项时，您将能够在 describe [table] extended 或 show table extended in [database] like '*' 的 Comment 字段中看到模型描述。

总是 schema，而不是 database

Apache Spark 使用 "schema" 和 "database" 互为同义词。dbt 将 database 理解为高于 schema 的层级。因此，您在运行 dbt-glue 时不应使用或设置 database 作为节点配置或在目标配置文件中。

如果您想控制 dbt 将模型物化到的模式/数据库，请仅使用 schema 配置和 generate_schema_name 宏。

注意事项

支持的功能

大多数 dbt Core 功能都受支持，但一些功能仅适用于 Apache Hudi。

Apache Hudi 专用功能

1. 通过 unique_key 而不是 partition_by 进行增量模型更新（请参阅 merge 策略）

某些在核心适配器上可用的 dbt 功能在 Glue 上尚不支持

1. 持久化 列级别描述作为数据库注释
2. 快照

---

有关 dbt 的更多信息

• 阅读 dbt 简介。
• 阅读 dbt 观点。
• 加入 dbt 社区。

---

安全

更多信息请参阅 CONTRIBUTING。

许可证

本项目采用Apache-2.0许可证。