Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Uso de dbt con Amazon MWAA

Con Amazon MWAA, puede usar dbt (herramienta de creación de datos) y PostgreSQL para crear y ejecutar flujos de trabajo de transformación de datos. En los pasos siguientes, añada las dependencias necesarias mediante un script de inicio y suba un proyecto de dbt de muestra al bucket de Amazon S3 de su entorno. A continuación, utilice un DAG de muestra para comprobar que Amazon MWAA ha instalado las dependencias. Por último, utilícelo BashOperator para ejecutar el proyecto dbt.

Versión

Puede usar el ejemplo de código de esta página con Apache Airflow v2 en Python 3.12 en el sitio web de Python y Apache Airflow v3 en Python 3.12 en el sitio web de Python.

Requisitos previos

Antes de poder completar los pasos siguientes, necesitará lo siguiente:

Dependencias

Para usar Amazon MWAA con dbt, agregue el siguiente script de inicio a su entorno. Para obtener más información, consulte Cómo usar un script de inicio 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 -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

En las siguientes secciones, suba el directorio de su proyecto de dbt a Amazon S3 y ejecute un DAG que valide si Amazon MWAA ha instalado correctamente las dependencias de dbt requeridas.

Carga de un proyecto de dbt en Amazon S3

Para poder utilizar un proyecto dbt con su entorno Amazon MWAA, cargue todo el directorio del proyecto a la carpeta de dags de su entorno. Cuando el entorno se actualice, Amazon MWAA descargará el directorio dbt en la carpeta local de usr/local/airflow/dags/.

Uso de un DAG para verificar la instalación de la dependencia de dbt

El siguiente DAG utiliza un comando BashOperator y un comando bash para verificar si Amazon MWAA ha instalado correctamente las dependencias de dbt especificadas en el script de inicio.

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

Realice los siguientes pasos para ver los registros de tareas y comprobar que dbt y sus dependencias se hayan instalado.

Cree y cargue un dbt profiles.yml

Para conectarse a la base de datos de destino, dbt necesita un archivo. profiles.yml El DAG de la siguiente sección pasa--profiles-dir /tmp/dbt, por lo que dbt lo busca profiles.yml directamente dentro del /tmp/dbt directorio. Esta es la dbt carpeta que ha subido a Amazon S3.

El nombre del perfil profiles.yml debe coincidir con el profile: valor definido en el proyecto inicialdbt_project.yml. Los usuarios default del proyecto dbt starter.

Uso de un DAG para ejecutar un proyecto dbt

El siguiente DAG utiliza un BashOperator para copiar los proyectos dbt que ha cargado en Amazon S3 desde el directorio local usr/local/airflow/dags/ al directorio accesible para escritura /tmp y, a continuación, ejecuta el proyecto dbt. Los comandos de bash asumen un proyecto dbt inicial titulado “dbt-starter-project”. Modifique el nombre del directorio de acuerdo con el nombre del directorio de su proyecto.

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