Skill de Pipeline de Dados

Construindo e revisando pipelines ETL/ELT - uma Skill de Agente para Python 3.14, pandas 2.2+ e Polars 1.x.

O Que Esta Skill Faz

Estrutura pipelines em lote: conectores de origem, camada de validação, funções de transformação, saídas particionadas (Parquet/Delta), ganchos de orquestração (Airflow/dbt/Cron) e verificações de qualidade de dados com IDs de execução reproduzíveis.

Quando Invocar

• Nova ingestão noturna de API SaaS ou exportação de data warehouse
• Revisão de PR na lógica de transformação pandas/Polars
• Adicionando validação após incidente de pico de nulos em produção
• Migrando protótipo de notebook para job agendado

Entradas

Saídas

• Layout do módulo do pipeline (extract.py, transform.py, load.py, validate.py)
• Dependências de pyproject.toml fixadas com uv
• Stubs de verificação Pandera ou Great Expectations
• Convenção de caminho de partição s3://bucket/dataset/dt=YYYY-MM-DD/
• Estratégia de carregamento idempotente (chaves de mesclagem, partição de sobrescrita)
• Comandos de verificação e asserções de contagem de linhas

Guardrails

1. Sem coerção silenciosa de dtype - esquemas explícitos na leitura e escrita.
2. Validar antes de transformar - falhar rapidamente em quebras de contrato.
3. Cargas idempotentes - novas execuções não devem duplicar linhas de fatos.
4. Segredos a partir de env - nunca codifique tokens de API em notebooks commitados no git.
5. Registrar run_id em cada estágio - correlacionar logs de extração/transformação/carga.
6. Não usar pandas em produção sem testes - mínimo de pytest em funções puras de transformação.

Receita

uv run python -m pipeline.run --date 2026-07-08 --env staging
uv run pytest tests/test_transform.py -q
uv run python -m pipeline.validate --date 2026-07-08

Exemplo de Trabalho (estágio de validação)

import pandera.pandas as pa
from pandera.typing import Series
 
 
class OrdersSchema(pa.DataFrameModel):
    order_id: Series[str] = pa.Field(unique=True)
    amount: Series[float] = pa.Field(ge=0)
    status: Series[str] = pa.Field(isin=["pending", "paid", "refunded"])
 
 
def validate_orders(df):
    return OrdersSchema.validate(df, lazy=True)

FAQs

Relacionados

• Noções Básicas de Engenharia de Dados - padrões
• Validação e Qualidade de Dados - verificações
• Airflow - orquestração
• Noções Básicas de pandas - referência de transformação

Versões do Stack: Esta página foi escrita para Python 3.14.0 (3.14 estável, 3.13 de manutenção), 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+.