本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

# 在 Amazon MWAA 中使用 dbt
<a name="samples-dbt"></a>

本主题演示了如何在 Amazon MWAA 中使用 dbt 和 Postgres。在以下步骤中，您将所需的依赖项添加到 `requirements.txt` 中，并将示例 dbt 项目上传到环境的 Amazon S3 存储桶。然后，您将使用示例 DAG 来验证 Amazon MWAA 是否已安装依赖项，最后使用 `BashOperator` 来运行 dbt 项目。

**Topics**
+ [版本](#samples-dbt-version)
+ [先决条件](#samples-dbt-prereqs)
+ [依赖项](#samples-dbt-dependencies)
+ [将 dbt 项目上传到 Amazon S3](#samples-dbt-upload-project)
+ [使用 DAG 验证 dbt 依赖项的安装](#samples-dbt-test-dependencies)
+ [使用 DAG 来运行 dbt 项目](#samples-dbt-run-project)

## 版本
<a name="samples-dbt-version"></a>

您可以在 [Python 3.10](https://peps.python.org/pep-0619/) 中将本页上的代码示例与 **Apache Airflow v2** 一起使用，在 [Python 3.11](https://peps.python.org/pep-0664/) 中与 **Apache Airflow v3** 一起使用。

## 先决条件
<a name="samples-dbt-prereqs"></a>

在完成以下步骤之前，您需要具备以下条件：
+ 使用 Apache Airflow v2.2.2 的 [Amazon MWAA 环境](get-started.md)。此示例已编写，并使用 v2.2.2 进行了测试。您可能需要修改示例以与其他 Apache Airflow 版本一起使用。
+ dbt 项目示例。要开始在 Amazon MWAA 中使用 dbt，你可以创建一个分支并从 dbt-labs 存储库中克隆 [dbt 入门项目](https://github.com/dbt-labs/dbt-starter-project)。 GitHub 

## 依赖项
<a name="samples-dbt-dependencies"></a>

要将 Amazon MWAA 与 dbt 配合使用，请将以下启动脚本添加到环境中。要了解更多信息，请参阅[在 Amazon MWAA 中使用启动脚本](using-startup-script.md)。

```
#!/bin/bash
			
  if [[ "${MWAA_AIRFLOW_COMPONENT}" != "worker" ]]
    then
      exit 0
  fi
			
  echo "------------------------------"
  echo "Installing virtual Python env"
  echo "------------------------------"
			
  pip3 install --upgrade pip
			
  echo "Current Python version:"
  python3 --version 
  echo "..."
			
  sudo pip3 install --user virtualenv
  sudo mkdir python3-virtualenv
  cd python3-virtualenv
  sudo python3 -m venv dbt-env
  sudo chmod -R 777 *
			
  echo "------------------------------"
  echo "Activating venv in"
  $DBT_ENV_PATH
	  		echo "------------------------------"
			
  source dbt-env/bin/activate
  pip3 list
			
  echo "------------------------------"
  echo "Installing libraries..."
  echo "------------------------------"
			
  # do not use sudo, as it will install outside the venv
  pip3 install dbt-redshift==1.6.1 dbt-postgres==1.6.1
			
  echo "------------------------------"
  echo "Venv libraries..."
  echo "------------------------------"
			
  pip3 list
  dbt --version
			
  echo "------------------------------"
  echo "Deactivating venv..."
  echo "------------------------------"
			
  deactivate
```

在以下章节中，您可将 dbt 项目目录上传到 Amazon S3 并运行 DAG 来验证 Amazon MWAA 是否已成功安装所需的 dbt 依赖项。

## 将 dbt 项目上传到 Amazon S3
<a name="samples-dbt-upload-project"></a>

为了能够在 Amazon MWAA 环境中使用 dbt 项目，您可以将整个项目目录上传到环境的 `dags` 文件夹中。当环境更新时，Amazon MWAA 会将 dbt 目录下载到本地 `usr/local/airflow/dags/` 文件夹。

**要将 dbt 项目上传到 Amazon S3，请执行以下操作**

1. 导航到您克隆 dbt 入门项目的目录。

1. 运行以下 Amazon S3 AWS CLI 命令，使用`--recursive`参数以递归方式将项目内容复制到您的环境`dags`文件夹。该命令会创建一个名为 `dbt` 的子目录，您可以将其用于所有 dbt 项目。如果子目录已经存在，则项目文件将被复制到现有目录中，并且不会创建新目录。该命令还会为该特定入门项目在 `dbt` 目录中创建一个子目录。

   ```
   aws s3 cp {{dbt-starter-project}} s3://{{amzn-s3-demo-bucket}}/dags/dbt/{{dbt-starter-project}} --recursive
   ```

   您可以为项目子目录使用不同的名称，以便在 `dbt` 父目录中组织多个 dbt 项目。

## 使用 DAG 验证 dbt 依赖项的安装
<a name="samples-dbt-test-dependencies"></a>

以下 DAG 使用 `BashOperator` 和 bash 命令来验证 Amazon MWAA 是否已成功安装 `requirements.txt` 中指定的 dbt 依赖项。

```
from airflow import DAG
			from airflow.operators.bash_operator import BashOperator
			from airflow.utils.dates import days_ago
			
			with DAG(dag_id="dbt-installation-test", schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
			cli_command = BashOperator(
			task_id="bash_command",
			bash_command="/usr/local/airflow/python3-virtualenv/dbt-env/bin/dbt --version"
			)
```

执行以下操作以访问任务日志并验证是否已安装 dbt 及其依赖项。

1. 导航到 Amazon MWAA 控制台，然后从可用环境列表中选择**打开 Airflow UI**。

1. 在 Apache Airflow UI 上，从列表中找到 `dbt-installation-test` DAG，然后在 `Last Run` 列中选择日期以打开上一个成功的任务。

1. 使用**图表视图**，选择 `bash_command` 任务以打开任务实例的详细信息。

1. 选择**日志**来打开任务日志，然后验证日志是否成功列出了我们在 `requirements.txt` 中指定的 dbt 版本。

## 使用 DAG 来运行 dbt 项目
<a name="samples-dbt-run-project"></a>

以下 DAG 使用 `BashOperator` 将您从本地 `usr/local/airflow/dags/` 目录上传到 Amazon S3 的 dbt 项目复制到可写入的 `/tmp` 目录，然后运行 dbt 项目。bash 命令假设一个名为 `dbt-starter-project` 的 入门 dbt 项目。根据您项目目录的名称修改目录名称。

```
from airflow import DAG
			from airflow.operators.bash_operator import BashOperator
			from airflow.utils.dates import days_ago
			
			import os
			
			DAG_ID = os.path.basename(__file__).replace(".py", "")
			
			# assumes all files are in a subfolder of DAGs called dbt
			
			with DAG(dag_id=DAG_ID, schedule_interval=None, catchup=False, start_date=days_ago(1)) as dag:
			cli_command = BashOperator(
			task_id="bash_command",
			bash_command="source /usr/local/airflow/python3-virtualenv/dbt-env/bin/activate;\
			cp -R /usr/local/airflow/dags/dbt /tmp;\
			echo 'listing project files:';\
			ls -R /tmp;\
			cd /tmp/dbt/mwaa_dbt_test_project;\
			/usr/local/airflow/python3-virtualenv/dbt-env/bin/dbt run --project-dir /tmp/dbt/mwaa_dbt_test_project --profiles-dir ..;\
			cat /tmp/dbt_logs/dbt.log;\
			rm -rf /tmp/dbt/mwaa_dbt_test_project"
			)
```