🚗 Pipeline Concessionária

📋 Descrição do Projeto

Pipeline completo de ETL para análise de dados de uma concessionária de veículos. O projeto extrai dados operacionais de vendas, clientes, produtos e categorias de um banco de produção PostgreSQL, realiza transformações com dbt seguindo arquitetura de camadas, e disponibiliza os dados em um Data Warehouse para visualização em Power BI.

Toda a orquestração é feita via Apache Airflow, permitindo agendamento automático, monitoramento e retry de falhas.

🏗️ Arquitetura do Pipeline

┌─────────────────┐
│  PostgreSQL     │
│  (Produção)     │
│  - vendas       │
│  - clientes     │
│  - produtos     │
└────────┬────────┘
         │ Extract
         ▼
┌─────────────────┐
│  Apache Airflow │
│  DAG Pipeline   │
│  - Extração     │
│  - Validação    │
│  - Trigger dbt  │
└────────┬────────┘
         │ Load
         ▼
┌─────────────────┐
│  PostgreSQL     │
│  (Data Warehouse)│
│  - staging      │
└────────┬────────┘
         │ Transform
         ▼
┌─────────────────┐
│      dbt        │
│  - staging      │
│  - dimensions   │
│  - analytics    │
└────────┬────────┘
         │
         ▼
┌─────────────────┐
│   Power BI      │
│  Dashboards     │
│  - Vendas       │
│  - Clientes     │
│  - Produtos     │
└─────────────────┘
        

Fluxo de Dados

1. Extração: Airflow conecta ao banco de produção e extrai dados via SQL
2. Carga: Dados brutos são carregados na camada staging do DW
3. Transformação: dbt aplica transformações em 3 camadas (staging → dimensions → analytics)
4. Visualização: Power BI consome as tabelas finais para dashboards

🛠️ Stack Utilizada

💻 Exemplo de Código

DAG do Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data_team',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'concessionaria_etl',
    default_args=default_args,
    description='Pipeline ETL Concessionária',
    schedule_interval='0 2 * * *',  # Diariamente às 2h
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:
    
    extract_vendas = PostgresOperator(
        task_id='extract_vendas',
        postgres_conn_id='prod_db',
        sql='sql/extract_vendas.sql'
    )
    
    run_dbt = BashOperator(
        task_id='run_dbt',
        bash_command='cd /opt/dbt && dbt run --models analytics'
    )
    
    extract_vendas >> run_dbt

Modelo dbt (Exemplo)

-- models/analytics/fact_vendas.sql
{{ config(
    materialized='table',
    schema='analytics'
) }}

WITH vendas_enriched AS (
    SELECT 
        v.venda_id,
        v.data_venda,
        v.valor_total,
        c.nome_cliente,
        c.cidade,
        p.modelo,
        p.categoria
    FROM {{ ref('stg_vendas') }} v
    LEFT JOIN {{ ref('dim_clientes') }} c ON v.cliente_id = c.cliente_id
    LEFT JOIN {{ ref('dim_produtos') }} p ON v.produto_id = p.produto_id
)

SELECT * FROM vendas_enriched

🎯 Desafios e Soluções

📊 Resultados e Métricas

Impacto no Negócio

• Dashboards atualizados diariamente, permitindo tomada de decisão baseada em dados atuais
• Redução de 85% no tempo de geração de relatórios gerenciais
• Visibilidade em tempo real de vendas, estoque e performance de produtos
• Base sólida para expansão futura (ML, previsões, etc.)

💡 Principais Aprendizados

• ✓ Importância de testes de qualidade de dados desde o início do projeto
• ✓ Benefícios da extração incremental para reduzir carga no banco de origem
• ✓ Valor do monitoramento proativo e alertas em pipelines de produção
• ✓ Como estruturar transformações em camadas lógicas (staging → dimensions → analytics)