Habilidad de Pipeline de Datos

Construcción y revisión de pipelines ETL/ELT - una Habilidad de Agente para Python 3.14, pandas 2.2+ y Polars 1.x.

Qué Hace Esta Habilidad

Proporciona un andamiaje para pipelines por lotes: conectores de origen, capa de validación, funciones de transformación, salidas particionadas (Parquet/Delta) y ganchos de orquestación (Airflow/dbt/Cron), y comprobaciones de calidad de datos con IDs de ejecución reproducibles.

Cuándo Invocar

• Nueva ingesta nocturna desde una API SaaS o exportación de un almacén de datos
• Revisión de PR en la lógica de transformación de pandas Polars
• Añadir validación tras un incidente de pico de nulos en producción
• Migrar un prototipo de notebook a un trabajo programado

Entradas

Salidas

• Diseño del módulo del pipeline (extract.py, transform.py, load.py, validate.py)
• Dependencias de pyproject.toml fijadas con uv
• Stubs de comprobación de Pandera o Great Expectations
• Convención de ruta de partición s3://bucket/dataset/dt=YYYY-MM-DD/
• Estrategia de carga idempotente (claves de fusión, sobrescritura de partición)
• Comandos de verificación y aserciones de recuento de filas

Barreras de Protección

1. Sin coerción silenciosa de tipos de datos - esquemas explícitos en lectura y escritura.
2. Validar antes de transformar - fallar rápidamente ante rupturas de contrato.
3. Cargas idempotentes - las reejecuciones no deben duplicar filas de hechos.
4. Secretos desde el entorno - nunca codificar tokens de API en notebooks que se confirmen en git.
5. Registrar run_id en cada etapa - correlacionar logs de extracción/transformación/carga.
6. No usar pandas en producción sin pruebas - mínimo pytest en funciones puras de transformación.

Receta

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

Ejemplo de Trabajo (etapa de validación)

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)

Preguntas Frecuentes

Relacionado

• Conceptos Básicos de Ingeniería de Datos - patrones
• Validación y Calidad de Datos - comprobaciones
• Airflow - orquestación
• Conceptos Básicos de pandas - referencia de transformación

Versiones de Stack: Esta página fue escrita para Python 3.14.0 (estable 3.14, mantenimiento 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+ y uv 0.6+.