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
| Entrada | Por qué |
|---|---|
| Tipo de origen | API, CSV de S3, réplica de Postgres, streaming |
| Elección del motor | ADR de pandas vs Polars |
| Grano | Eventos a nivel de fila vs agregados diarios |
| SLA | Ventana de frescura y tiempo de inactividad permitido |
| Columnas PII | Requisitos de enmascaramiento/hash |
Salidas
- Diseño del módulo del pipeline (
extract.py,transform.py,load.py,validate.py) - Dependencias de
pyproject.tomlfijadas 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
- Sin coerción silenciosa de tipos de datos - esquemas explícitos en lectura y escritura.
- Validar antes de transformar - fallar rápidamente ante rupturas de contrato.
- Cargas idempotentes - las reejecuciones no deben duplicar filas de hechos.
- Secretos desde el entorno - nunca codificar tokens de API en notebooks que se confirmen en git.
- Registrar
run_iden cada etapa - correlacionar logs de extracción/transformación/carga. - 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-08Ejemplo 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
¿pandas o Polars por defecto?
Sigue el ADR del equipo - Polars para transformaciones grandes en un solo nodo; pandas cuando las integraciones del ecosistema dominan.
¿Airflow o cron?
cron + systemd para un solo trabajo; Airflow cuando importan dependencias DAG, backfill y SLAs.
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+.