As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Como usar DBT com o Amazon MWAA

Com o Amazon MWAA, você pode usar o dbt (ferramenta de criação de dados) e o PostgreSQL para criar e executar fluxos de trabalho de transformação de dados. Nas etapas a seguir, adicione as dependências necessárias usando um script de inicialização e faça o upload de um projeto dbt de amostra para o bucket Amazon S3 do seu ambiente. Em seguida, use um exemplo de DAG para verificar se o Amazon MWAA instalou as dependências. Finalmente, use o BashOperator para executar o projeto dbt.

Versão

Você pode usar o exemplo de código nesta página com o Apache Airflow v2 em Python 3.12 no site do Python e o Apache Airflow v3 no Python 3.12 no site do Python. https://peps.python.org/pep-0693/

Pré-requisitos

Antes de concluir as etapas a seguir, você precisa do seguinte:

Dependências

Para usar o Amazon MWAA com DBT, adicione o script de inicialização a seguir ao ambiente. Para saber mais, consulte Usar um script de startup com o 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

Nas seções a seguir, faça o upload do diretório do seu projeto dbt para o Amazon S3 e execute um DAG que valida se o Amazon MWAA instalou com êxito as dependências dbt necessárias.

Faça o upload de um projeto de DBT para o Amazon S3

Para poder usar um projeto de DBT com seu ambiente Amazon MWAA, é possível fazer o upload todo o diretório do projeto na pasta dags do seu ambiente. Quando o ambiente é atualizado, o Amazon MWAA baixa o diretório DBT para a pasta local usr/local/airflow/dags/.

Use um DAG para verificar a instalação da dependência de DBT

O DAG a seguir usa um comando bash BashOperator e um comando para verificar se o Amazon MWAA instalou com êxito as dependências dbt especificadas no script de inicialização.

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

Faça o seguinte para acessar os logs de tarefas e verificar se o dbt e as dependências dele foram instalados.

Crie e faça upload de um dbt profiles.yml

Para se conectar ao seu banco de dados de destino, o dbt requer um profiles.yml arquivo. O DAG na próxima seção é aprovado--profiles-dir /tmp/dbt, então o dbt procura profiles.yml diretamente dentro do /tmp/dbt diretório. Essa é a dbt pasta que você enviou para o Amazon S3.

O nome do perfil em profiles.yml deve corresponder ao profile: valor definido no projeto inicial. dbt_project.yml O projeto dbt starter usa. default

Use um DAG para executar um projeto de DBT

O DAG a seguir usa um código BashOperator para copiar os projetos de DBT que você fez o upload para o Amazon S3 do diretório local usr/local/airflow/dags/ para o diretório acessível para gravação /tmp e, em seguida, executa o projeto de DBT. Os comandos bash pressupõem um projeto inicial de DBT intitulado dbt-starter-project. Modifique o nome do diretório de acordo com o nome do diretório do seu projeto.

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