dbt mit Amazon verwenden MWAA

Dieses Thema zeigt, wie Sie dbt und Postgres mit Amazon verwenden können. MWAA In den folgenden Schritten fügen Sie Ihrem requirements.txt die erforderlichen Abhängigkeiten hinzu und laden ein DBT-Beispielprojekt in den Amazon S3 S3-Bucket Ihrer Umgebung hoch. Anschließend überprüfen Sie anhand eines BeispielsDAG, ob Amazon die Abhängigkeiten installiert MWAA hat, und verwenden das schließlich, BashOperator um das dbt-Projekt auszuführen.

Version

• Sie können das Codebeispiel auf dieser Seite mit Apache Airflow v2 in Python 3.10 verwenden.

Voraussetzungen

Bevor Sie die folgenden Schritte ausführen können, benötigen Sie Folgendes:

• Eine MWAAAmazon-Umgebung, die Apache Airflow v2.2.2 verwendet. Dieses Beispiel wurde mit Version 2.2.2 geschrieben und getestet. Möglicherweise müssen Sie das Beispiel ändern, um es mit anderen Apache Airflow-Versionen verwenden zu können.

• Ein DBT-Beispielprojekt. Um mit der Verwendung von dbt mit Amazon zu beginnenMWAA, können Sie einen Fork erstellen und das dbt-Starter-Projekt aus dem GitHub dbt-labs-Repository klonen.

Abhängigkeiten

Um Amazon MWAA mit dbt zu verwenden, fügen Sie Ihrer Umgebung das folgende Startskript hinzu. Weitere Informationen finden Sie unter Verwenden eines Startskripts mit 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

In den folgenden Abschnitten laden Sie Ihr dbt-Projektverzeichnis auf Amazon S3 hoch und führen einen aus, der überprüftDAG, ob Amazon die erforderlichen dbt-Abhängigkeiten erfolgreich installiert MWAA hat.

Laden Sie ein dbt-Projekt auf Amazon S3 hoch

Um ein dbt-Projekt mit Ihrer MWAA Amazon-Umgebung verwenden zu können, können Sie das gesamte Projektverzeichnis in den dags Ordner Ihrer Umgebung hochladen. Wenn die Umgebung aktualisiert wird, MWAA lädt Amazon das dbt-Verzeichnis in den lokalen usr/local/airflow/dags/ Ordner herunter.

Um ein dbt-Projekt auf Amazon S3 hochzuladen

1. Navigieren Sie zu dem Verzeichnis, in dem Sie das dbt-Starterprojekt geklont haben.

2. Führen Sie den folgenden Amazon S3 AWS CLI S3-Befehl aus, um den Inhalt des Projekts mithilfe des --recursive Parameters rekursiv in den dags Ordner Ihrer Umgebung zu kopieren. Der Befehl erstellt ein Unterverzeichnis namensdbt, das Sie für all Ihre DBT-Projekte verwenden können. Wenn das Unterverzeichnis bereits existiert, werden die Projektdateien in das bestehende Verzeichnis kopiert, und es wird kein neues Verzeichnis erstellt. Der Befehl erstellt auch ein Unterverzeichnis innerhalb des dbt Verzeichnisses für dieses spezielle Starterprojekt.

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

   Sie können verschiedene Namen für Projektunterverzeichnisse verwenden, um mehrere DBT-Projekte innerhalb des übergeordneten Verzeichnisses zu organisieren. dbt

Verwenden Sie aDAG, um die Installation der DBT-Abhängigkeit zu überprüfen

Im Folgenden werden ein BashOperator und ein Bash-Befehl DAG verwendet, um zu überprüfen, ob Amazon die in angegebenen dbt-Abhängigkeiten erfolgreich installiert MWAA hat. 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""
    )

Gehen Sie wie folgt vor, um die Task-Logs einzusehen und zu überprüfen, ob dbt und die zugehörigen Abhängigkeiten installiert wurden.

1. Navigieren Sie zur MWAA Amazon-Konsole und wählen Sie dann Open Airflow UI aus der Liste der verfügbaren Umgebungen aus.

2. Suchen Sie auf der Apache Airflow-Benutzeroberfläche in der Liste nach und wählen Sie dann das Datum unter der Last Run Spalte dbt-installation-test DAG aus, an dem die letzte erfolgreiche Aufgabe geöffnet werden soll.

3. Wählen Sie in der Diagrammansicht die bash_command Aufgabe aus, um die Details der Aufgabeninstanz zu öffnen.

4. Wählen Sie Log, um die Task-Logs zu öffnen, und vergewissern Sie sich dann, dass die Logs die von uns angegebene DBT-Version erfolgreich auflisten. requirements.txt

Verwenden Sie aDAG, um ein DBT-Projekt auszuführen

Im Folgenden DAG werden die dbt-ProjekteBashOperator, die Sie auf Amazon S3 hochgeladen haben, vom lokalen usr/local/airflow/dags/ Verzeichnis in das Verzeichnis mit Schreibzugriff /tmp kopiert und anschließend das dbt-Projekt ausgeführt. Die Bash-Befehle gehen von einem DBT-Startprojekt mit dem Titel aus. dbt-starter-project Ändern Sie den Verzeichnisnamen entsprechend dem Namen Ihres Projektverzeichnisses.

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