View a markdown version of this page

搭配 Amazon MWAA 使用 dbt - Amazon Managed Workflows for Apache Airflow
版本先決條件相依性將 dbt 專案上傳至 Amazon S3使用 DAG 驗證 dbt 相依性安裝建立和上傳 dbt profile.yml使用 DAG 執行 dbt 專案

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

搭配 Amazon MWAA 使用 dbt

透過 Amazon MWAA，您可以使用 dbt （資料建置工具） 和 PostgreSQL 來建置和執行資料轉換工作流程。在下列步驟中，使用啟動指令碼新增必要的相依性，並將範例 dbt 專案上傳至您環境的 Amazon S3 儲存貯體。然後，使用範例 DAG 來驗證 Amazon MWAA 已安裝相依性。最後，使用 BashOperator執行 dbt 專案。

版本

您可以在 Python 網站上將此頁面上的程式碼範例與 Python 3.12 中的 Apache Airflow v2 和 Python 網站上的 Python 3.12 中的 Apache Airflow v3 搭配使用。

先決條件

在您可以完成下列步驟之前，您需要下列項目：

  • 使用 Apache Airflow 2.11.2 版的 Amazon MWAA 環境。此範例已撰寫，並使用 v2.11.2 進行測試。您可能需要修改範例，以搭配其他 Apache Airflow 版本使用。

  • 範例 dbt 專案。若要開始使用 dbt 搭配 Amazon MWAA，您可以建立分支，並從 dbt-labs GitHub 儲存庫複製 dbt 入門專案。 GitHub

相依性

若要將 Amazon MWAA 與 dbt 搭配使用，請將下列啟動指令碼新增至您的環境。若要進一步了解，請參閱搭配 Amazon MWAA 使用啟動指令碼。

#!/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 -p /usr/local/airflow/python3-virtualenv
cd /usr/local/airflow/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-core==1.9.4 dbt-redshift==1.9.1 dbt-postgres==1.9.0

echo "------------------------------"
echo "Venv libraries..."
echo "------------------------------"

pip3 list
dbt --version

echo "------------------------------"
echo "Deactivating venv..."
echo "------------------------------"

deactivate
設定 DBT_ENV_PATH 變數

您可以在啟動指令碼$DBT_ENV_PATH中設定 ，或在 Amazon MWAA 環境中將其設定為 Airflow 組態。

在下列各節中，將您的 dbt 專案目錄上傳至 Amazon S3，並執行 DAG，以驗證 Amazon MWAA 是否已成功安裝所需的 dbt 相依性。

將 dbt 專案上傳至 Amazon S3

若要將 dbt 專案與 Amazon MWAA 環境搭配使用，您可以將整個專案目錄上傳至環境的dags資料夾。當環境更新時，Amazon MWAA 會將 dbt 目錄下載至本機usr/local/airflow/dags/資料夾。

將 dbt 專案上傳至 Amazon S3

  1. 導覽至您複製 dbt 入門專案的目錄。

  2. 執行下列 Amazon S3 AWS CLI command，使用 --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 相依性安裝

下列 DAG 使用 BashOperator和 bash 命令來驗證 Amazon MWAA 是否已成功安裝啟動指令碼中指定的 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 主控台，然後從可用環境清單中選擇 Open Airflow UI。

  2. 在 Apache Airflow UI 上，從清單中尋找 dbt-installation-test DAG，然後在Last Run欄中選擇日期以開啟最後一個成功的任務。

  3. 使用圖形檢視，選擇bash_command任務以開啟任務執行個體詳細資訊。

  4. 選擇日誌以開啟任務日誌，然後驗證日誌是否成功列出啟動指令碼中指定的 dbt 版本。

建立和上傳 dbt profile.yml

若要連線到您的目標資料庫，dbt 需要 profiles.yml 檔案。下一節的 DAG 會傳遞 --profiles-dir /tmp/dbt，因此 dbt 會profiles.yml直接在 /tmp/dbt 目錄中尋找 。這是您上傳到 Amazon S3 的dbt資料夾。

中的設定檔名稱profiles.yml必須符合入門專案 中定義的profile:值dbt_project.yml。dbt 入門專案使用 default。

建立和上傳 profile.yml

  1. 在您複製入門專案的目錄中，profiles.yml使用資料庫連線詳細資訊建立名為 的檔案。

    default:
  target: dev
  outputs:
    dev:
      type: postgres
      host: your-db-endpoint.region.rds.amazonaws.com
      port: 5432
      user: your_db_user
      password: your_db_password
      dbname: your_database
      schema: your_schema
      threads: 4

  2. 將檔案上傳至您環境 DAGs資料夾中的 dbt 子目錄。

    aws s3 cp profiles.yml s3://amzn-s3-demo-bucket/dags/dbt/profiles.yml
保護資料庫登入資料

為了避免儲存純文字登入資料，請使用 dbt env_var()函數參考秘密。透過 Amazon MWAA 環境變數或 AWS Secrets Manager提供值，例如 password: "{{ env_var('DBT_PASSWORD') }}"。此外，請確定您的 Amazon MWAA VPC 安全群組可讓您的工作者連線到已設定連接埠上的資料庫。

使用 DAG 執行 dbt 專案

下列 DAG 使用 BashOperator將您上傳至 Amazon S3 的 dbt 專案從本機usr/local/airflow/dags/目錄複製到可寫入/tmp目錄，然後執行 dbt 專案。bash 命令會假設名為 的入門 dbt 專案dbt-starter-project。根據專案目錄的名稱修改目錄名稱。

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/dbt-starter-project;\
        /usr/local/airflow/python3-virtualenv/dbt-env/bin/dbt run --project-dir /tmp/dbt/dbt-starter-project --profiles-dir /tmp/dbt;\
        cat /tmp/dbt/dbt-starter-project/logs/dbt.log;\
        rm -rf /tmp/dbt/dbt-starter-project"
    )