Laboratório: Pipeline ETL automatizado de BI com Apache Airflow, PostgreSQL e Pandas

Duração estimada

120 min

Passos

11

Validações

4

Versões da documentação consultada

Apache Airflow 2.8.1 · PostgreSQL 16 · Python 3.10 · Pandas 2.2.2

Topologia

Três componentes principais: 1) Apache Airflow (192.168.56.10) orquestra o pipeline; 2) Banco PostgreSQL de origem (192.168.56.11) fornece dados brutos; 3) Banco PostgreSQL de destino (192.168.56.12) recebe os dados tratados. O Airflow conecta-se a ambos os bancos via rede privada.

      +--------------------+
      |   Apache Airflow  |
      | 192.168.56.10     |
      +--------+----------+
               |
    +----------+----------+
    |                     |
+---+---+           +-----+---+
| Origem|           | Destino |
|Postgre|           |Postgre  |
|.11    |           |.12      |
+-------+           +---------+

Equipamentos e software

Roteiro

1. Preparar as VMs e rede

   Crie três VMs (ou containers) em uma rede privada (ex: 192.168.56.0/24). Garanta comunicação entre elas.

   O que você deve ver: Ping entre as VMs funciona; acesso SSH disponível.

2. Instalar PostgreSQL nas VMs de origem e destino

   Instale e configure o PostgreSQL em ambas as VMs. Defina senhas e permita conexões pela rede privada.

   sudo apt update && sudo apt install postgresql
   sudo -u postgres psql -c "ALTER USER postgres PASSWORD 'postgres';"
   # Edite /etc/postgresql/16/main/pg_hba.conf e /etc/postgresql/16/main/postgresql.conf conforme a documentação.Copiar

   O que você deve ver: Serviço PostgreSQL ativo em ambas as VMs, aceitando conexões remotas.

3. Criar bancos e tabelas de teste

   Crie um banco 'bi_origem' na origem e 'bi_destino' no destino. Crie tabelas simples para simular dados brutos e tratados.

   sudo -u postgres createdb bi_origem
   sudo -u postgres createdb bi_destino
   # Na origem:
   sudo -u postgres psql -d bi_origem -c "CREATE TABLE vendas (id serial PRIMARY KEY, produto text, valor numeric, data_venda date);"
   # No destino:
   sudo -u postgres psql -d bi_destino -c "CREATE TABLE vendas_tratadas (id serial PRIMARY KEY, produto text, valor numeric, data_venda date, valor_com_imposto numeric);"Copiar

   O que você deve ver: Tabelas criadas em ambos os bancos.

4. Popular dados de teste na origem

   Insira alguns registros fictícios na tabela de origem para simular dados reais.

   sudo -u postgres psql -d bi_origem -c "INSERT INTO vendas (produto, valor, data_venda) VALUES ('Fibra 100M', 100, '2024-06-01'), ('Fibra 200M', 180, '2024-06-02');"Copiar

   O que você deve ver: Dados disponíveis na tabela vendas da origem.

5. Instalar Apache Airflow na VM dedicada

   Instale o Apache Airflow com suporte a PostgreSQL e Pandas. Use ambiente virtual Python para isolar dependências.

   sudo apt install python3-venv
   python3 -m venv airflow_venv
   source airflow_venv/bin/activate
   pip install "apache-airflow[postgres,pandas]==2.8.1" --constraint "https://raw.githubusercontent.com/apache/airflow/constraints-2.8.1/constraints-3.10.txt"Copiar

   O que você deve ver: Airflow e dependências instalados sem erros.

6. Inicializar e configurar o Airflow

   Inicialize o banco de dados do Airflow e crie um usuário admin. Configure o Airflow para rodar localmente.

   export AIRFLOW_HOME=~/airflow
   airflow db init
   airflow users create --username admin --password admin --firstname Admin --lastname User --role Admin --email admin@lab.localCopiar

   O que você deve ver: Airflow inicializado e usuário admin criado.

7. Configurar conexões do Airflow com os bancos

   Adicione conexões no Airflow para os dois bancos PostgreSQL usando a interface web ou CLI.

   airflow connections add 'pg_origem' --conn-uri 'postgresql+psycopg2://postgres:postgres@192.168.56.11:5432/bi_origem'
   airflow connections add 'pg_destino' --conn-uri 'postgresql+psycopg2://postgres:postgres@192.168.56.12:5432/bi_destino'Copiar

   O que você deve ver: Conexões pg_origem e pg_destino criadas no Airflow.

8. Criar DAG ETL no Airflow

   Implemente um DAG que: 1) extrai dados do banco origem, 2) transforma com Pandas (adiciona coluna de imposto), 3) carrega no banco destino.

   mkdir -p ~/airflow/dags
   touch ~/airflow/dags/etl_bi.py
   # Exemplo de DAG:
   # Consulte https://airflow.apache.org/docs/apache-airflow/stable/howto/operator/python.html para detalhes.
   # Veja trecho abaixo:
   
   from airflow import DAG
   from airflow.providers.postgres.hooks.postgres import PostgresHook
   from airflow.operators.python import PythonOperator
   from datetime import datetime
   import pandas as pd
   
   def extrair(**context):
       hook = PostgresHook(postgres_conn_id='pg_origem')
       df = hook.get_pandas_df('SELECT * FROM vendas;')
       df.to_csv('/tmp/vendas.csv', index=False)
   
   def transformar(**context):
       df = pd.read_csv('/tmp/vendas.csv')
       df['valor_com_imposto'] = df['valor'] * 1.1
       df.to_csv('/tmp/vendas_tratadas.csv', index=False)
   
   def carregar(**context):
       hook = PostgresHook(postgres_conn_id='pg_destino')
       df = pd.read_csv('/tmp/vendas_tratadas.csv')
       for _, row in df.iterrows():
           hook.run(
               "INSERT INTO vendas_tratadas (produto, valor, data_venda, valor_com_imposto) VALUES (%s, %s, %s, %s)",
               parameters=(row['produto'], row['valor'], row['data_venda'], row['valor_com_imposto'])
           )
   
   default_args = {'start_date': datetime(2024, 6, 1)}
   with DAG('etl_bi', schedule_interval=None, default_args=default_args, catchup=False) as dag:
       t1 = PythonOperator(task_id='extrair', python_callable=extrair)
       t2 = PythonOperator(task_id='transformar', python_callable=transformar)
       t3 = PythonOperator(task_id='carregar', python_callable=carregar)
       t1 >> t2 >> t3Copiar

   O que você deve ver: Arquivo etl_bi.py criado em ~/airflow/dags.

9. Iniciar o Airflow Scheduler e Webserver

   Suba o scheduler e o webserver do Airflow para monitorar e executar o DAG.

   airflow scheduler &
   airflow webserver --port 8080 &Copiar

   O que você deve ver: Airflow acessível em http://localhost:8080.

10. Executar o DAG manualmente

    No painel do Airflow, ative e execute o DAG etl_bi. Observe o status das tarefas.

    O que você deve ver: Execução do DAG com três tarefas: extrair, transformar, carregar.

11. Verificar dados no banco de destino

    Confirme que os dados tratados chegaram corretamente ao banco de destino.

    sudo -u postgres psql -d bi_destino -c "SELECT * FROM vendas_tratadas;"Copiar

    O que você deve ver: Registros com coluna valor_com_imposto presentes na tabela vendas_tratadas.

Como saber que funcionou

Troubleshooting

Segurança e hardening

• Nunca use senhas padrão em ambientes de produção
• Restrinja o acesso às portas do PostgreSQL apenas à rede privada
• Evite salvar dados sensíveis em arquivos temporários sem criptografia
• Configure roles e permissões mínimas necessárias no PostgreSQL
• Desative usuários e conexões não utilizados após o laboratório

Checklist final

• VMs comunicando entre si
• PostgreSQL instalado e acessível nas duas VMs
• Tabelas criadas e populadas
• Airflow instalado e configurado
• Conexões do Airflow testadas
• DAG criado e executado com sucesso
• Dados transformados presentes no destino