El viaje de un dato: de su origen hasta una decisión de negocio

Ya conoces los ingredientes: tipos de datos, formatos (CSV, JSON, Parquet) y dónde se almacenan (archivos, bases de datos). Ahora toca entender cómo se conecta todo: el VIAJE que hace un dato desde que nace hasta que alguien toma una decisión con él. Este viaje es lo que llamamos "pipeline de datos" — y es literalmente el core de tu futuro trabajo.

Piensa en el agua de tu grifo. El agua nace en un río o un embalse (origen). Pasa por una potabilizadora donde la filtran y desinfectan (transformación). Se almacena en depósitos (almacenamiento). Viaja por tuberías hasta tu casa (transporte). Y finalmente la usas para cocinar o beber (consumo). Si cualquier eslabón falla — contaminación en el río, tubería rota, depósito vacío — no tienes agua limpia. Un pipeline de datos funciona EXACTAMENTE igual.

### Las tres fases: Extraer, Transformar, Cargar

El patrón más básico y universal de un pipeline se llama ETL: Extract (extraer datos del origen), Transform (limpiar, validar, enriquecer) y Load (cargar en el destino). Ya lo hiciste en la Skill de Python sin saberlo: cuando leíste un CSV, limpiaste los datos y escribiste el resultado en otro archivo, estabas haciendo ETL. La diferencia en producción es que esto pasa automáticamente, cada día, con millones de registros y sin que nadie lo toque.

### Ejemplo concreto: el pipeline de una tienda online

La tienda "PixelShop" vende electrónica online. Cada noche necesita un reporte de ventas por producto para que el equipo de compras sepa qué reponer. Veamos el viaje de los datos:

1. 01.ORIGEN: la app de la tienda guarda cada pedido en un archivo JSON diario (pedidos_2026-01-15.json).
2. 02.EXTRAER: nuestro script Python lee ese archivo JSON.
3. 03.TRANSFORMAR: limpiamos datos (quitar pedidos cancelados, convertir moneda, validar importes), enriquecemos (calcular importe total = precio × cantidad) y agregamos (sumar por producto).
4. 04.CARGAR: escribimos el resultado en un archivo Parquet organizado por fecha.
5. 05.CONSUMIR: el equipo de compras consulta un dashboard que lee esos Parquets y muestra qué reponer.

1import json
2import csv
3from datetime import date, timedelta
4from collections import defaultdict
5
6# === EXTRACT: leer datos del origen ===
7# (simulamos el JSON de pedidos del día)
8pedidos_raw = [
9    {"id": 1, "producto": "Mouse", "precio": 25.0, "cantidad": 2, "estado": "completado"},
10    {"id": 2, "producto": "Teclado", "precio": 60.0, "cantidad": 1, "estado": "completado"},
11    {"id": 3, "producto": "Mouse", "precio": 25.0, "cantidad": 1, "estado": "cancelado"},
12    {"id": 4, "producto": "Monitor", "precio": 350.0, "cantidad": 1, "estado": "completado"},
13    {"id": 5, "producto": "Teclado", "precio": 60.0, "cantidad": 3, "estado": "completado"},
14    {"id": 6, "producto": "Webcam", "precio": 80.0, "cantidad": 1, "estado": "completado"},
15]
16print(f"EXTRACT: {len(pedidos_raw)} pedidos leídos")
17
18# === TRANSFORM: limpiar + enriquecer + agregar ===
19# 1. Filtrar: solo pedidos completados
20pedidos_validos = [p for p in pedidos_raw if p["estado"] == "completado"]
21print(f"TRANSFORM: {len(pedidos_validos)} pedidos válidos (sin cancelados)")
22
23# 2. Enriquecer: calcular importe total
24for p in pedidos_validos:
25    p["importe_total"] = p["precio"] * p["cantidad"]
26
27# 3. Agregar: sumar por producto
28ventas_por_producto = defaultdict(lambda: {"unidades": 0, "importe": 0})
29for p in pedidos_validos:
30    ventas_por_producto[p["producto"]]["unidades"] += p["cantidad"]
31    ventas_por_producto[p["producto"]]["importe"] += p["importe_total"]
32
33# === LOAD: guardar resultado ===
34hoy = date.today().isoformat()
35with open(f"reporte_ventas_{hoy}.csv", "w", newline="") as f:
36    writer = csv.writer(f)
37    writer.writerow(["producto", "unidades_vendidas", "importe_total", "fecha"])
38    for prod, datos in sorted(ventas_por_producto.items(), key=lambda x: -x[1]["importe"]):
39        writer.writerow([prod, datos["unidades"], datos["importe"], hoy])
40
41print(f"LOAD: reporte guardado como reporte_ventas_{hoy}.csv")
42print("\n=== RESULTADO (lo que ve el equipo de compras) ===")
43for prod, datos in sorted(ventas_por_producto.items(), key=lambda x: -x[1]["importe"]):
44    print(f"  {prod:10} | {datos['unidades']} uds | {datos['importe']:>8.2f}€")

### Las zonas de los datos: raw, procesado y listo

En un sistema profesional, los datos no saltan directamente del origen al reporte. Pasan por zonas, como estaciones en un tren. La idea es simple: mantener el dato original intacto (por si necesitas reprocesar) y tener versiones cada vez más limpias y útiles.

• ▹Zona Raw (bruto): el dato tal cual llega del origen. Sin tocar. El JSON original, el CSV del proveedor. Es tu seguro: si algo sale mal en la transformación, siempre puedes volver aquí.
• ▹Zona Procesada (limpio): el dato ya limpio, validado y en formato estándar. Has quitado duplicados, convertido tipos, validado rangos. Listo para transformar.
• ▹Zona Final (listo para uso): el dato agregado, enriquecido y en el formato que el consumidor necesita. El reporte, la tabla del dashboard, el archivo para el equipo de compras.

En tu proyecto de la Skill anterior (el script que procesaba un CSV de ventas) ya hiciste esto intuitivamente: leíste el archivo original (raw), lo limpiaste (procesado) y generaste un resultado (final). En sistemas profesionales, estas zonas son carpetas en un disco, carpetas en la nube o tablas diferentes en una base de datos. La idea es idéntica.

### Cuándo corren los pipelines: batch vs continuo

Hay dos grandes modos de ejecutar pipelines. El modo BATCH (lote) significa que el pipeline se ejecuta a intervalos: cada hora, cada día, cada semana. Es como el correo postal: se acumula durante el día y se reparte una vez. El modo CONTINUO (streaming) significa que los datos se procesan según llegan, en tiempo real o casi real. Es como los mensajes de WhatsApp: instantáneos.

El 90% de los pipelines en la industria son batch. El reporte de ventas diario, la carga de datos del proveedor semanal, la actualización del dashboard cada hora. Streaming se usa cuando necesitas reaccionar en segundos: detección de fraude, alertas de temperatura, actualizaciones de ubicación de repartidores. Empezarás con batch — es más simple, más predecible y suficiente para la mayoría de casos.

### ¿Qué puede salir mal? (spoiler: todo)

Los pipelines parecen simples en teoría: leer, transformar, guardar. ¿Qué puede fallar? TODO. Y lo hará. A las 3 de la mañana de un domingo:

• ▹El origen no responde: la API del proveedor está caída, el archivo no llegó, el servidor está lleno.
• ▹Los datos cambiaron de formato: el proveedor añadió una columna nueva sin avisarte, o cambió "precio" por "price".
• ▹Datos corruptos: una fila tiene caracteres extraños, un número es negativo cuando no debería, un campo fecha tiene formato incorrecto.
• ▹El destino está lleno: el disco se llenó, la base de datos no acepta más conexiones.
• ▹Error parcial: procesaste 999.000 de 1.000.000 registros y el 999.001 falló. ¿Guardas los 999.000 o tiras todo?
• ▹Duplicados: el pipeline se ejecutó dos veces (porque alguien lo relanzó sin querer). ¿Datos duplicados en el destino?

### Resumen

1. 01.Un pipeline de datos es una secuencia automática: Extraer → Transformar → Cargar (ETL).
2. 02.Los datos pasan por zonas: raw (original intacto), procesado (limpio) y final (listo para uso).
3. 03.Los pipelines pueden ser batch (periódicos) o streaming (en tiempo real). El 90% son batch.
4. 04.NUNCA borres los datos raw: son tu seguro ante errores.
5. 05.Los pipelines fallan constantemente. Diseña pensando en el fallo: reintentos, alertas, validaciones.