dbt (ferramenta de transformação de dados)

dbt

dbt permite que engenheiros de analytics transformem dados escrevendo instruções SQL SELECT. Ele gerencia materialização (tabelas, views, incremental), testes, documentação e rastreamento de linhagem.

Instalação

# Instalar dbt com adaptador PostgreSQL
pip install dbt-postgres

# Ou com outros adaptadores
pip install dbt-bigquery
pip install dbt-snowflake

# Inicializar um novo projeto
dbt init my_project
cd my_project

Estrutura do Projeto

my_project/
├── dbt_project.yml      # Configuração do projeto
├── profiles.yml         # Perfis de conexão (geralmente em ~/.dbt/)
├── models/
│   ├── staging/         # Limpeza de dados brutos
│   │   ├── _staging.yml # Schema + testes para modelos de staging
│   │   ├── stg_users.sql
│   │   └── stg_orders.sql
│   └── marts/           # Lógica de negócios
│       ├── _marts.yml
│       └── fct_revenue.sql
├── tests/               # Testes de dados personalizados
├── macros/              # Macros SQL reutilizáveis
└── seeds/               # Arquivos CSV para carregar

Configuração

# dbt_project.yml: Configuração do projeto
name: my_project
version: '1.0.0'
profile: my_project

models:
  my_project:
    staging:
      +materialized: view
      +schema: staging
    marts:
      +materialized: table
      +schema: analytics

# profiles.yml: Conexão com banco de dados (~/.dbt/profiles.yml)
my_project:
  target: dev
  outputs:
    dev:
      type: postgres
      host: localhost
      port: 5432
      user: analyst
      password: "{{ env_var('DBT_PASSWORD') }}"
      dbname: analytics
      schema: dev
      threads: 4
    prod:
      type: postgres
      host: prod-db.example.com
      port: 5432
      user: dbt_prod
      password: "{{ env_var('DBT_PROD_PASSWORD') }}"
      dbname: analytics
      schema: public
      threads: 8

Modelos de Staging

-- models/staging/stg_users.sql: Limpeza de dados brutos de usuários
WITH source AS (
    SELECT * FROM {{ source('raw', 'users') }}
),

cleaned AS (
    SELECT
        id AS user_id,
        LOWER(TRIM(email)) AS email,
        name,
        created_at::timestamp AS signed_up_at,
        CASE WHEN status = 'active' THEN TRUE ELSE FALSE END AS is_active
    FROM source
    WHERE email IS NOT NULL
)

SELECT * FROM cleaned

-- models/staging/stg_orders.sql: Limpeza de dados brutos de pedidos
SELECT
    id AS order_id,
    user_id,
    amount_cents / 100.0 AS amount,
    status,
    created_at::timestamp AS ordered_at
FROM {{ source('raw', 'orders') }}
WHERE status != 'test'

Modelos de Mart

-- models/marts/fct_revenue.sql: Tabela de fatos de receita
{{
    config(
        materialized='incremental',
        unique_key='order_date',
        on_schema_change='sync_all_columns'
    )
}}

WITH orders AS (
    SELECT * FROM {{ ref('stg_orders') }}
    {% if is_incremental() %}
    WHERE ordered_at > (SELECT MAX(order_date) FROM {{ this }})
    {% endif %}
),

daily AS (
    SELECT
        DATE_TRUNC('day', ordered_at)::date AS order_date,
        COUNT(*) AS total_orders,
        COUNT(DISTINCT user_id) AS unique_customers,
        SUM(amount) AS total_revenue,
        AVG(amount) AS avg_order_value
    FROM orders
    WHERE status = 'completed'
    GROUP BY 1
)

SELECT * FROM daily

Schema e Testes

# models/staging/_staging.yml: Definir fontes, colunas e testes
version: 2

sources:
  - name: raw
    schema: public
    tables:
      - name: users
        loaded_at_field: created_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}
      - name: orders

models:
  - name: stg_users
    description: Dados limpos de usuários
    columns:
      - name: user_id
        tests: [unique, not_null]
      - name: email
        tests: [unique, not_null]

  - name: stg_orders
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: ['pending', 'completed', 'cancelled', 'refunded']

Comandos CLI

# commands.sh: Comandos comuns da CLI do dbt
# Executar todos os modelos
dbt run

# Executar modelo específico e suas dependências upstream
dbt run --select +fct_revenue

# Executar testes
dbt test

# Gerar e servir documentação
dbt docs generate
dbt docs serve --port 8081

# Verificar frescor das fontes
dbt source freshness

# Build completo (run + test + snapshot)
dbt build

# Executar contra produção
dbt run --target prod

Macros

-- macros/cents_to_dollars.sql: Macro reutilizável para conversão de moeda
{% macro cents_to_dollars(column_name) %}
    ({{ column_name }} / 100.0)::numeric(10,2)
{% endmacro %}

-- Uso em um modelo: SELECT {{ cents_to_dollars('amount_cents') }} AS amount