Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Menggunakan dbt dengan Amazon MWAA

Dengan Amazon MWAA, Anda dapat menggunakan dbt (alat pembuatan data) dan PostgreSQL untuk membangun dan menjalankan alur kerja transformasi data. Pada langkah-langkah berikut, tambahkan dependensi yang diperlukan menggunakan skrip startup, dan unggah contoh proyek dbt ke bucket Amazon S3 lingkungan Anda. Kemudian, gunakan contoh DAG untuk memverifikasi bahwa Amazon MWAA telah menginstal dependensi. Terakhir, gunakan BashOperator untuk menjalankan proyek dbt.

Versi

Anda dapat menggunakan contoh kode di halaman ini dengan Apache Airflow v2 di Python 3.12 di situs web Python dan Apache Airflow v3 di Python 3.12 di situs web Python.

Prasyarat

Sebelum Anda dapat menyelesaikan langkah-langkah berikut, Anda memerlukan yang berikut:

Dependensi

Untuk menggunakan Amazon MWAA dengan dbt, tambahkan skrip startup berikut ke lingkungan Anda. Untuk mempelajari selengkapnya, lihat Menggunakan skrip startup dengan 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

Di bagian berikut, unggah direktori proyek dbt Anda ke Amazon S3 dan jalankan DAG yang memvalidasi apakah Amazon MWAA telah berhasil menginstal dependensi dbt yang diperlukan.

Unggah proyek dbt ke Amazon S3

Untuk dapat menggunakan proyek dbt dengan lingkungan Amazon MWAA Anda, Anda dapat mengunggah seluruh direktori proyek ke folder lingkungan Anda. dags Saat lingkungan diperbarui, Amazon MWAA mengunduh direktori dbt ke folder lokalusr/local/airflow/dags/.

Gunakan DAG untuk memverifikasi instalasi ketergantungan dbt

DAG berikut menggunakan perintah a BashOperator dan bash untuk memverifikasi apakah Amazon MWAA telah berhasil menginstal dependensi dbt yang ditentukan dalam skrip startup.

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

Lakukan hal berikut untuk mengakses log tugas dan memverifikasi bahwa dbt dan dependensinya telah diinstal.

Buat dan unggah profil.yml dbt

Untuk terhubung ke database target Anda, dbt memerlukan profiles.yml file. DAG di bagian berikutnya berlalu--profiles-dir /tmp/dbt, jadi dbt mencari profiles.yml langsung di dalam /tmp/dbt direktori. Ini adalah dbt folder yang Anda unggah ke Amazon S3.

Nama profil di profiles.yml harus sesuai dengan profile: nilai yang ditentukan dalam proyek pemuladbt_project.yml. Proyek starter dbt menggunakandefault.

Gunakan DAG untuk menjalankan proyek dbt

DAG berikut menggunakan a BashOperator untuk menyalin proyek dbt yang Anda unggah ke Amazon S3 dari usr/local/airflow/dags/ direktori lokal ke direktori yang dapat diakses penulisan/tmp, lalu menjalankan proyek dbt. Perintah bash mengasumsikan proyek dbt starter berjudul. dbt-starter-project Ubah nama direktori sesuai dengan nama direktori proyek Anda.

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