翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

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 で使用できます。 https://peps.python.org/pep-0693/

前提条件

次の手順を実行する前に、以下が必要です。

依存関係

dbt で Amazon MWAA を使用するには、次のスタートアップスクリプトを環境に追加します。詳細については、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 プロジェクトディレクトリを Amazon S3 にアップロードし、Amazon MWAA が必要な dbt 依存関係を正常にインストールしたかどうかを検証する DAG を実行します。

DBT プロジェクトを Amazon S3 にアップロードする

Amazon MWAA 環境で dbt プロジェクトを使用できるようにするには、プロジェクトディレクトリ全体を環境の dags フォルダにアップロードできます。環境が更新されると、Amazon MWAA は dbt ディレクトリをローカル usr/local/airflow/dags/ フォルダにダウンロードします。

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 とその依存関係がインストールされていることを確認します。

dbt profiles.yml を作成してアップロードする

ターゲットデータベースに接続するには、dbt に profiles.yml ファイルが必要です。次のセクションの DAG は を渡すため--profiles-dir /tmp/dbt、dbt は /tmp/dbt ディレクトリ内でprofiles.yml直接 を探します。これは、Amazon S3 にアップロードしたdbtフォルダです。

のプロファイル名は、スタータープロジェクトの で定義されているprofile:値と一致するprofiles.yml必要がありますdbt_project.yml。dbt スタータープロジェクトは を使用しますdefault。

DAG を使用して dbt プロジェクトを実行します。

次の DAG は、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/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"
    )