dbt com Python

dbt gerencia transformações de warehouse como modelos versionados - principalmente SQL, com modelos Python onde os adaptadores permitem lógica estilo pandas/polars no runtime do warehouse.

Receita

Cartão de receita de referência rápida - pronto para copiar e colar.

# models/staging/stg_orders.sql
select
  order_id,
  cast(revenue as double) as revenue,
  region,
  ordered_at
from {{ source('raw', 'orders') }}
where order_id is not null

# models/marts/revenue_by_region.py (modelo Python dbt)
def model(dbt, session):
    df = dbt.ref("stg_orders").to_pandas()
    return df.groupby("region", observed=True)["revenue"].sum().reset_index()

Quando usar isso:

• Transformações de análise de propriedade de equipes de dados em SQL
• Necessidade de testes, documentação e linhagem em tabelas de warehouse
• Python apenas para recursos de estatística/ML que o SQL não consegue expressar de forma limpa
• CI executa dbt build em cada PR

Exemplo de Trabalho

-- models/staging/stg_orders.sql
with source as (
    select * from {{ source('ecommerce', 'orders') }}
),
cleaned as (
    select
        order_id,
        upper(trim(region)) as region,
        cast(revenue as numeric(18, 2)) as revenue,
        ordered_at::timestamp_tz as ordered_at
    from source
    where revenue >= 0
)
select * from cleaned

# models/staging/schema.yml
models:
  - name: stg_orders
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: revenue
        tests:
          - not_null
          - dbt_expectations.expect_column_values_to_be_between:
              min_value: 0

-- models/marts/fct_revenue_daily.sql
select
    date_trunc('day', ordered_at) as revenue_date,
    region,
    sum(revenue) as total_revenue,
    count(*) as order_count
from {{ ref('stg_orders') }}
group by 1, 2

O que isso demonstra:

• Grafo de dependência source e ref
• Limpeza de staging em SQL próximo ao motor do warehouse
• Testes YAML em chaves e limites de receita
• Agregação de mart como modelo downstream

Mergulho Profundo

Como Funciona

• dbt compila Jinja SQL, executa materializações (view, table, incremental).
• DAG a partir de ref() - dbt build executa testes após a construção do modelo.
• Modelos Python delegam para o runtime do adaptador (Snowflake Snowpark, BigQuery, etc.).
• Artefatos: manifest.json, catalog.json para ferramentas de linhagem.

Camadas

Notas Python

# Mantenha os modelos Python enxutos - o trabalho pesado ainda pertence ao SQL quando possível
def model(dbt, session):
    import pandas as pd
    orders = dbt.ref("stg_orders")
    pdf = orders.to_pandas() if hasattr(orders, "to_pandas") else orders
    # engenharia de recursos ...
    return pdf

Armadilhas

• Modelo Python no adaptador errado - nem todos os warehouses suportam modelos Python. Correção: verifique a documentação do adaptador; use UDF SQL como alternativa.
• Lógica duplicada em ferramentas de BI - métricas divergem dos marts dbt. Correção: exponha os grãos nos marts; BI apenas visualiza.
• Pular testes em staging - chaves incorretas se propagam. Correção: unique/not_null em chaves primárias no mínimo.
• Incremental de full-refresh por engano - reconstrução cara. Correção: documente incremental_strategy e chaves únicas.
• Nomes de banco de dados hard-coded - quebra entre ambientes. Correção: target perfis dev/prod via variáveis de ambiente.

Alternativas

FAQs

dbt debug && dbt build --select stg_orders+

tests:
  - relationships:
      to: ref('dim_customers')
      field: customer_id

Relacionado

• Validação e Qualidade de Dados - testes de expectativas
• Formatos de Arquivo - tabelas externas em Parquet
• Noções Básicas de Engenharia de Dados - ciclo de vida ETL
• PySpark - alternativa em escala de cluster

Versões da Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, e uv 0.6+.