Referência: Uma Plataforma de Dados / Sistema ETL

Um layout ETL de referência para equipes de dados em Python: ingestão de APIs e arquivos, transformação com Polars 1.x ou pandas 2.2+, carregamento para Snowflake/BigQuery/Postgres, orquestrado por Prefect ou Airflow com estágios idempotentes e SLAs de atualidade.

Receita

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

data-platform/
  flows/ingest_orders.py     # fluxo orquestrado
  transforms/orders.py       # funções puras
  loaders/warehouse.py       # COPY / MERGE
  contracts/orders.yaml      # esquema + SLAs
  tests/test_transforms.py

Quando usar isso:

• Construindo uma plataforma de análise em Python
• Substituindo scripts "cron spaghetti"
• RFC para limite de malha de dados (data mesh)

Exemplo de Trabalho

"""transforms/orders.py - transformação idempotente."""
from __future__ import annotations
 
import polars as pl
 
 
def normalize_orders(raw: pl.LazyFrame) -> pl.LazyFrame:
    return (
        raw.filter(pl.col("status").is_in(["paid", "refunded"]))
        .with_columns(
            pl.col("amount").cast(pl.Float64),
            pl.col("created_at").str.to_datetime(strict=False),
        )
        .select("order_id", "tenant_id", "amount", "status", "created_at")
    )

"""flows/ingest_orders.py - esboço de fluxo estilo Prefect."""
from __future__ import annotations
 
from datetime import datetime, timezone
 
import polars as pl
 
from loaders.warehouse import merge_into_warehouse
from transforms.orders import normalize_orders
 
WATERMARK_KEY = "orders_ingest"
 
 
def run_ingest(since: datetime) -> int:
    raw_path = f"s3://lake/raw/orders/{since.date()}.parquet"
    raw = pl.scan_parquet(raw_path)
    clean = normalize_orders(raw).collect()
    rows = merge_into_warehouse(clean, table="analytics.orders", key="order_id")
    return rows
 
 
if __name__ == "__main__":
    ts = datetime.now(timezone.utc)
    print(f"carregou {run_ingest(ts)} linhas")

# contracts/orders.yaml
name: orders
freshness_sla_hours: 6
columns:
  order_id: { type: string, required: true }
  amount: { type: float, min: 0 }

O que isso demonstra:

• Transformações são funções puras testadas sem o orquestrador
• Polars lazy para varreduras eficientes em memória
• Carregamento MERGE com chave para idempotência
• Contrato de dados documenta SLA e esquema

Análise Detalhada

Como Funciona

• Marca d'água (Watermark) - Armazena a marca d'água alta; seguro para reexecutar para a mesma janela.
• Orquestração - Retentativas, alertas, grafo de dependência (dimensões antes de fatos).
• Lake + warehouse - Dados brutos imutáveis; tabelas curadas no warehouse.
• Verificações de qualidade - Limites de contagem de linhas, taxas de nulos, validação de contrato.
• Linha do tempo (Lineage) - Documenta sistemas de origem por tabela no catálogo.

Modelo de Camadas

Notas de Python

# loaders/warehouse.py esboço
def merge_into_warehouse(df: pl.DataFrame, table: str, key: str) -> int:
    # escreve staging parquet → MERGE SQL
    return df.height

Armadilhas

• Append não idempotente - Linhas duplicadas na retentativa. Correção: MERGE na chave natural.
• Pandas gigante em memória - Falta de memória (OOM) no worker. Correção: Varredura lazy do Polars ou leitura em chunks.
• Cron sem observabilidade - Dados silenciosamente desatualizados. Correção: Alerta de atualidade em max(created_at).
• Mudança de esquema sem contrato - Quebra de ML downstream. Correção: Versionar o contrato YAML no PR.
• Compartilhamento de credenciais de DB em notebooks - Risco de vazamento. Correção: Apenas injeção de segredo pelo orquestrador.

Alternativas

FAQs

Relacionados

• Referência: Um Pipeline de ML de Produção - ML downstream
• dbt com Python - camada SQL
• Pipelines de Dados - padrões de orquestração
• Noções Básicas de Polars - motor de transformação
• Catálogo de Lições Aprendidas - erros comuns

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+.