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 리포지토리에서 dbt 스타터 프로젝트를 복제하면 됩니다. GitHub

의존성

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 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/ 폴더에 다운로드합니다.

Amazon S3에 dbt 프로젝트를 업로드하려면

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 프로젝트를 구성할 수 있습니다.

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

작업 로그를 보고 dbt 및 해당 종속성이 설치되었는지 확인하려면 다음을 수행합니다.

1. Amazon MWAA 콘솔로 이동한 다음 사용 가능한 환경 목록에서 Open Airflow UI를 선택합니다.

2. Apache Airflow UI의 목록에서 dbt-installation-test DAG를 찾은 다음 Last Run 열 아래에서 날짜를 선택하여 마지막으로 성공한 작업을 엽니다.

3. 그래프 보기를 사용하여 bash_command 작업을 선택하여 작업 인스턴스 세부 정보를 엽니다.

4. 로그를 선택하여 작업 로그를 연 다음, 로그에 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", "")

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