기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

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/ https://peps.python.org/pep-0693/

사전 조건

다음 단계를 완료하려면 먼저 다음이 필요합니다.

종속성

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 -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를 실행합니다.

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

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 profile.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"
    )