Amazon MWAA での dbt の使用

このトピックでは、Amazon MWAA で dbt と Postgres を使用する方法を示します。次のステップでは、必要な依存関係を requirements.txt に追加し、サンプルの dbt プロジェクトを環境の Amazon S3 バケットにアップロードします。次に、サンプル DAG を使用して Amazon MWAA が依存関係をインストールしたことを確認し、最後に BashOperator を使用して dbt プロジェクトを実行します。

バージョン

• このページのコード例は、「Python 3.10」で Apache Airflow v2 以上と共に使用可能です。

前提条件

次の手順を完了するには、以下のものが必要です。

• Apache Airflow v2.2.2 を使用する Amazon MWAA 環境。このサンプルは v2.2.2 で作成され、テストされています。他の Apache Airflow バージョンで使用するためには、サンプルを変更する必要がある場合があります。

• dbt プロジェクトのサンプル。Amazon MWAA で dbt を使い始めるには、フォークを作成し、dbt-labs GitHub リポジトリから dbt スタータープロジェクトをクローンすることができます。

依存関係

dbt で Amazon MWAA を使用するには、requirements.txt に次の依存関係を追加してください。詳細については、「Python 依存関係のインストール」を参照してください。

環境の更新が完了すると、Amazon MWAA は必要な dbt ライブラリと、psycopg2 などの追加の依存関係をインストールします。

注記

Apache Airflow v2.2.2 で提供されるデフォルトの制約ファイルには、このガイドで使用している dbt のバージョンではサポートされていない、jsonschema と矛盾するバージョンがあります。そのため、dbt で Amazon MWAA を使用する場合は、Apache Airflow 制約ファイルを Amazon S3 DAGs フォルダにダウンロードして変更し、--constraint /usr/local/airflow/dags/my-updated-constraint.txt などの requirements.txt ファイル内で参照するか、以下に示すように --constraint から requirements.txt を省略することができます。

json-rpc==1.13.0
minimal-snowplow-tracker==0.0.2
packaging==20.9
networkx==2.6.3 
mashumaro==2.5
sqlparse==0.4.2

logbook==1.5.3
agate==1.6.1
dbt-extractor==0.4.0

pyparsing==2.4.7 
msgpack==1.0.2
parsedatetime==2.6
pytimeparse==1.1.8
leather==0.3.4
pyyaml==5.4.1

# Airflow constraints are jsonschema==3.2.0
jsonschema==3.1.1
hologram==0.0.14
dbt-core==0.21.1

psycopg2-binary==2.8.6
dbt-postgres==0.21.1
dbt-redshift==0.21.1

以下のセクションでは、dbt プロジェクトディレクトリを Amazon S3 にアップロードし、Amazon MWAA が必要な dbt 依存関係を正常にインストールしたかどうかを検証する DAG を実行します。

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

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

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

1. dbt スタータープロジェクトをクローンしたディレクトリに移動します。

2. 次の Amazon S3 AWS CLI コマンドを実行して、dags パラメータを使用してプロジェクトの内容を再帰的に環境の --recursive にコピーします。このコマンドは、dbtと呼ばれるサブディレクトリを作成し、これをすべてのdbtプロジェクトに使用できます。サブディレクトリが既に存在する場合、プロジェクトファイルは既存のディレクトリにコピーされ、新しいディレクトリは作成されません。このコマンドは、この特定のスターターのために dbt ディレクトリ内にサブディレクトリも作成します。

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

   プロジェクトのサブディレクトリに異なる名前を使用して、親 dbt ディレクトリ内の複数の dbt プロジェクトを整理できます。

DAG を使用して dbt 依存関係のインストールを検証します。

次の DAGは、BashOperator を使用し、requirements.txt で指定されたdbtの依存関係を正常にインストールしたかどうかを Amazon MWAA が確認します。

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/.local/bin/dbt --version"
    )

以下を実行してタスクログを表示し、dbt とその依存関係がインストールされていることを確認します。

1. Amazon MWAA コンソールに移動し、使用可能な環境のリストから [Airflow UI を開く] を選択します。

2. Apache Airflow UI のリストから dbt-installation-test DAG を探し、Last Run列の下にある日付を選択して、最後に成功したタスクを開きます。

3. グラフビューを使用してタスクを選択し、bash_commandタスクインスタンスの詳細を開きます。

4. [Log] を選択してタスクログを開き、次に、requirements.txt で指定されたdbtのバージョンをログが正常にリストしていることを確認してください。

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", "")
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="cp -R /usr/local/airflow/dags/dbt /tmp;\
cd /tmp/dbt/dbt-starter-project;\
/usr/local/airflow/.local/bin/dbt  run --project-dir /tmp/dbt/dbt-starter-project/ --profiles-dir ..;\
cat /tmp/dbt_logs/dbt.log"
    )