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
| Entrada | Por quê |
|---|---|
| Tipo de origem | API, S3 CSV, réplica Postgres, streaming |
| Escolha do motor | ADR pandas vs Polars |
| Granularidade | Eventos em nível de linha vs. agregados diários |
| SLA | Janela de frescor e tempo de inatividade permitido |
| Colunas PII | Requisitos de mascaramento/hash |
Saídas
- Layout do módulo do pipeline (
extract.py,transform.py,load.py,validate.py) - Dependências de
pyproject.tomlfixadas 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
- Sem coerção silenciosa de dtype - esquemas explícitos na leitura e escrita.
- Validar antes de transformar - falhar rapidamente em quebras de contrato.
- Cargas idempotentes - novas execuções não devem duplicar linhas de fatos.
- Segredos a partir de env - nunca codifique tokens de API em notebooks commitados no git.
- Registrar run_id em cada estágio - correlacionar logs de extração/transformação/carga.
- 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-08Exemplo 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
pandas ou Polars por padrão?
Siga o ADR da equipe - Polars para transformações de nó único grandes; pandas quando integrações de ecossistema dominam.
Airflow ou cron?
cron + systemd para um único job; Airflow quando dependências de DAG, backfill e SLAs importam.
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+.