在 Amazon 上使用 dbt MWAA - Amazon Managed Workflows for Apache Airflow
版本必要條件相依性將 dbt 項目上傳到 Amazon S3使用 a DAG 來驗證 dbt 相依性安裝使用一個DAG來運行一個 dbt 項目

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

在 Amazon 上使用 dbt MWAA

本主題演示瞭如何將 dbt 和 Postgres 與 Amazon 一起使用。MWAA在以下步驟中，您將新增必要的相依性至您的requirements.txt，並將範例 dbt 專案上傳到環境的 Amazon S3 儲存貯體。然後，您將使用範例DAG來驗證 Amazon 是否MWAA已安裝相依性，最後使BashOperator用執行 dbt 專案。

版本

  • 您可以使用此頁面上的代碼示例與 Python 3.10 中的阿帕奇氣流 V2。

必要條件

您需要下列項目，才能完成下列步驟：

  • 使用阿帕奇氣流 v2.2.2 的 Amazon MWAA 環境。此範例已撰寫完成，並使用 v2.2.2 進行測試。您可能需要修改範例，才能與其他 Apache 氣流版本搭配使用。

  • 一個範例 dbt 專案。要開始在 Amazon 上使用 dbtMWAA，您可以創建一個分叉並從 db GitHub t-labs 存儲庫克隆 dbt 入門項目。

相依性

若要將 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 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，然後執行驗證 Amazon 是否MWAA已成功安裝所需的 dbt 相依性。DAG

將 dbt 項目上傳到 Amazon S3

為了能夠在 Amazon MWAA 環境中使用 dbt 項目，您可以將整個項目目錄上傳到環境的dags文件夾。當環境更新時，Amazon 將 dbt 目錄MWAA下載到本地文usr/local/airflow/dags/件夾。

將 dbt 專案上傳到 Amazon S3

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

  2. 執行下列 Amazon S3 AWS CLI 命令，以遞迴方式使用--recursive參數將專案內容複製到環境的dags資料夾。該命令創建一個名為的子目錄dbt，您可以將其用於所有 dbt 項目。如果子目錄已存在，則專案檔案會複製到現有目錄中，而不會建立新目錄。此指令也會在此特定入門專案的dbt目錄內建立子目錄。

    $ aws s3 cp dbt-starter-project s3://mwaa-bucket/dags/dbt/dbt-starter-project --recursive

    您可以為項目子目錄使用不同的名稱來組織父dbt目錄中的多個 dbt 項目。

使用 a DAG 來驗證 dbt 相依性安裝

以下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控台，然後從可用環境清單中選擇「開啟氣流使用者介面」。

  2. 在 Apache Airflow 使用者介面上，dbt-installation-testDAG從清單中找到，然後選擇Last Run欄下方的日期以開啟上次成功的工作。

  3. 使用「圖形檢視」，選擇要開啟bash_command作業執行環境詳細資訊的作業。

  4. 選擇 [記錄] 開啟工作記錄，然後驗證記錄檔是否成功列出我們在中requirements.txt指定的 dbt 版本。

使用一個DAG來運行一個 dbt 項目

以下DAG使用 a 將您上傳BashOperator到 Amazon S3 的 dbt 專案從本機usr/local/airflow/dags/目錄複製到可寫入存取的目/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"
    )