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

Topik ini menunjukkan bagaimana Anda dapat menggunakan dbt dan Postgres dengan Amazon MWAA. Pada langkah-langkah berikut, Anda akan menambahkan dependensi yang diperlukan ke perangkat Andarequirements.txt, dan mengunggah contoh proyek dbt ke bucket Amazon S3 lingkungan Anda. Kemudian, Anda akan menggunakan sampel DAG untuk memverifikasi bahwa Amazon MWAA telah menginstal dependensi, dan akhirnya menggunakan BashOperator untuk menjalankan proyek dbt.

Versi

Anda dapat menggunakan contoh kode pada halaman ini dengan Apache Airflow v2 di Python 3.10 dan Apache Airflowv3 di Python 3.11.

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 lebih lanjut, 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 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

Di bagian berikut, Anda akan mengunggah direktori proyek dbt Anda ke Amazon S3 dan menjalankan 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 lokal. usr/local/airflow/dags/

Gunakan DAG untuk memverifikasi instalasi ketergantungan dbt

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

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.

Gunakan DAG untuk menjalankan proyek dbt

DAG berikut menggunakan a BashOperator untuk menyalin proyek dbt yang Anda unggah ke Amazon S3 dari direktori usr/local/airflow/dags/ lokal ke direktori /tmp yang dapat diakses tulis, lalu jalankan 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/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"
			)