Amazon MWAA에서 dbt 사용

이 주제에서는 Amazon MWAA으로 dbt 및 Postgres를 사용하는 방법을 보여줍니다. 다음 단계에서는 필수 종속성을 requirements.txt에 추가하고 샘플 dbt 프로젝트를 환경의 Amazon S3 버킷에 업로드합니다. 그런 다음 샘플 DAG를 사용하여 Amazon MWAA가 종속성을 설치했는지 확인하고 마지막으로 BashOperator를 사용하여 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를 실행합니다.

Amazon S3에 dbt 프로젝트를 업로드합니다.

Amazon MWAA 환경에서 dbt 프로젝트를 사용할 수 있으려면 전체 프로젝트 디렉터리를 환경 dags 폴더에 업로드하면 됩니다. 환경이 업데이트되면 Amazon MWAA는 dbt 디렉터리를 로컬 usr/local/airflow/dags/ 폴더에 다운로드합니다.

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 및 해당 종속성이 설치되었는지 확인하려면 다음을 수행합니다.

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/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"
    )