Referencia: Una Plataforma de Datos / Sistema ETL

Un diseño de referencia de ETL para equipos de datos de Python: ingesta desde APIs y archivos, transformación con Polars 1.x o pandas 2.2+, carga a Snowflake/BigQuery/Postgres, orquestado por Prefect o Airflow con etapas idempotentes y SLAs de frescura.

Receta

Tarjeta de receta de referencia rápida - lista para copiar y pegar.

data-platform/
  flows/ingest_orders.py     # flujo orquestado
  transforms/orders.py       # funciones puras
  loaders/warehouse.py       # COPIAR / MERGE
  contracts/orders.yaml      # esquema + SLAs
  tests/test_transforms.py

Cuándo usar esto:

• Construcción de una plataforma de análisis en Python
• Reemplazo de scripts de "espagueti" de cron
• RFC para el límite de la malla de datos

Ejemplo de Trabajo

"""transforms/orders.py - transformación 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 - esquema de flujo 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"cargados {run_ingest(ts)} filas")

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

Lo que esto demuestra:

• Las transformaciones son funciones puras probadas sin orquestador
• Polars perezoso para escaneos eficientes en memoria
• Carga MERGE con clave para idempotencia
• El contrato de datos documenta el SLA y el esquema

Inmersión Profunda

Cómo Funciona

• Marca de agua - Almacena la marca de agua alta; reintento seguro para la misma ventana.
• Orquestación - Reintentos, alertas, grafo de dependencias (dimensiones antes que hechos).
• Lago + almacén de datos - Crudo inmutable; tablas curadas en el almacén de datos.
• Comprobaciones de calidad - Límites de recuento de filas, tasas de nulos, validación de contratos.
• Linaje - Documenta sistemas de origen por tabla en el catálogo.

Modelo de Capas

Notas de Python

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

Trampas

• Anexión no idempotente - Filas duplicadas en reintento. Solución: MERGE sobre clave natural.
• Pandas gigantes en memoria - OOM en el trabajador. Solución: Escaneo perezoso de Polars o lectura por fragmentos.
• Cron sin observabilidad - Datos obsoletos silenciosos. Solución: Alerta de frescura en max(created_at).
• Cambio de esquema sin contrato - ML descendente falla. Solución: Contrato de versión YAML en PR.
• Compartir credenciales de BD en notebooks - Riesgo de fuga. Solución: Inyección de secretos solo del orquestador.

Alternativas

Preguntas Frecuentes

Relacionado

• Referencia: Un Pipeline de ML de Producción - ML descendente
• dbt con Python - capa SQL
• Pipelines de Datos - patrones de orquestación
• Conceptos Básicos de Polars - motor de transformación
• Catálogo de Lecciones Aprendidas - errores comunes

Versiones de la pila: 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+.