Usare dbt con Amazon MWAA

Questo argomento dimostra come utilizzare dbt e Postgres con Amazon MWAA. Nei passaggi seguenti, aggiungerai le dipendenze richieste al tuo requirements.txt e caricherai un progetto dbt di esempio nel bucket Amazon S3 del tuo ambiente. Quindi, utilizzerai un DAG di esempio per verificare che Amazon MWAA abbia installato le dipendenze e infine utilizzerai il BashOperator per eseguire il progetto dbt.

Versione

• Puoi usare l'esempio di codice in questa pagina con Apache Airflow v2 e versioni successive in Python 3.10.

Prerequisiti

Prima di poter completare i seguenti passaggi, avrai bisogno di quanto segue:

• Un ambiente Amazon MWAA che utilizza Apache Airflow v2.2.2. Questo esempio è stato scritto e testato con la versione 2.2.2. Potrebbe essere necessario modificare l'esempio per utilizzarlo con altre versioni di Apache Airflow.

• Un esempio di progetto dbt. Per iniziare a usare dbt con Amazon MWAA, puoi creare un fork e clonare il progetto dbt starter dal repository dbt-labs. GitHub

Dipendenze

Per usare Amazon MWAA con dbt, aggiungi il seguente script di avvio al tuo ambiente. Per ulteriori informazioni, consulta Usare uno script di avvio con 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

Nelle sezioni seguenti, caricherai la tua directory di progetto dbt su Amazon S3 ed eseguirai un DAG che convalida se Amazon MWAA ha installato correttamente le dipendenze dbt richieste.

Caricare un progetto dbt su Amazon S3

Per poter utilizzare un progetto dbt con il tuo ambiente Amazon MWAA, puoi caricare l'intera directory del progetto nella cartella del tuo ambiente. dags Quando l'ambiente si aggiorna, Amazon MWAA scarica la directory dbt nella cartella locale. usr/local/airflow/dags/

Per caricare un progetto dbt su Amazon S3

1. Passa alla directory in cui hai clonato il progetto dbt starter.

2. Esegui il seguente AWS CLI comando Amazon S3 per copiare in modo ricorsivo il contenuto del progetto dags nella cartella del tuo ambiente utilizzando il parametro. --recursive Il comando crea una sottodirectory chiamata dbt che puoi usare per tutti i tuoi progetti dbt. Se la sottodirectory esiste già, i file di progetto vengono copiati nella directory esistente e non viene creata una nuova directory. Il comando crea anche una sottodirectory all'interno della dbt directory per questo specifico progetto iniziale.

   $ aws s3 cp dbt-starter-project s3://mwaa-bucket/dags/dbt/dbt-starter-project --recursive

   È possibile utilizzare nomi diversi per le sottodirectory del progetto per organizzare più progetti dbt all'interno della directory principale. dbt

Utilizzate un DAG per verificare l'installazione della dipendenza da dbt

Il seguente DAG utilizza un comando BashOperator e un comando bash per verificare se Amazon MWAA ha installato correttamente le dipendenze dbt specificate in. 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/.local/bin/dbt --version"
    )

Effettua le seguenti operazioni per visualizzare i log delle attività e verificare che dbt e le sue dipendenze siano state installate.

1. Accedi alla console Amazon MWAA, quindi scegli Open Airflow UI dall'elenco degli ambienti disponibili.

2. Nell'interfaccia utente di Apache Airflow, individua il dbt-installation-test DAG dall'elenco, quindi scegli la data sotto la Last Run colonna per aprire l'ultima attività riuscita.

3. Utilizzando Graph View, scegliete l'bash_commandattività per aprire i dettagli dell'istanza dell'operazione.

4. Scegliete Log per aprire i log delle attività, quindi verificate che i log elencino correttamente la versione dbt in cui è stata specificata. requirements.txt

Usa un DAG per eseguire un progetto dbt

Il seguente DAG utilizza BashOperator a per copiare i progetti dbt caricati su Amazon S3 dalla directory usr/local/airflow/dags/ locale alla directory /tmp accessibile in scrittura, quindi esegue il progetto dbt. I comandi bash presuppongono un progetto dbt iniziale intitolato. dbt-starter-project Modifica il nome della directory in base al nome della directory del tuo progetto.

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