Serie MLOps y Producción · de la ingeniería de datos al monitoreo.
Módulo I — Fundamentos y arquitectura de sistemas de IA
La sesión 1 dejó dos pipelines y un manifiesto. Esta sesión escribe el primero: el pipeline de datos. Su nombre industrial es ETL —Extract, Transform, Load— y su promesa no es “limpiar el Excel”. Es esta:
Dada una fuente y una marca de agua, entrego una tabla con columnas \(C\), tipos acordados y un hash \(h\). No puntúo. No elijo \(f\).
Tres confusiones que este cuaderno se niega a alimentar:
ETL no es calidad. Tipar una fecha y filtrar un precio negativo son transformaciones estructurales. Decidir si un nulo es MCAR y si un outlier se imputa es la sesión 3.
ETL no es feature engineering.cantidad * precio como total de la línea es grano de negocio. El \(\log(1+x)\) y el target encoding son la sesión 4.
ETL no es orquestación. Aquí hay funciones y carpetas. El DAG, los reintentos y Kahn son la sesión 7.
El hilo: extraer sin mutar el origen, transformar con contrato, cargar de forma idempotente, y dejar linaje.
Objetivos de aprendizaje
Al finalizar, el estudiante estará en capacidad de:
Definir ETL como contrato de movimiento y distinguirlo de ELT, de la calidad y de \(T(x)\).
Extraer desde archivo, tabla y “API” (contrato JSON) sin alterar el origen.
Justificar extracción completa frente a incremental (marca de agua) y detectar datos tardíos.
Aplicar transformaciones estructurales: tipos, husos, grano, uniones —sin imputar ni encodear para el modelo.
Cargar en modos overwrite / append / merge, con capas bronce y plata, de forma idempotente.
Dejar hash y linaje de la foto, listos para que la sesión 3 juzgue calidad.
1) ETL como patrón, no como herramienta
Las siglas ordenan tres responsabilidades:
Fase
Pregunta que cierra
Fallo típico si se omite
E Extraer
¿De dónde y hasta cuándo?
Se “baja el Excel” a mano y nadie sabe de qué lunes es.
T Transformar
¿Qué esquema sale?
Cada analista tipa las fechas distinto.
L Cargar
¿Dónde vive la foto y se puede repetir?
Se pisa el crudo o se duplica el lote.
Surgió con los almacenes analíticos (años 70–80) porque los sistemas transaccionales (OLTP) no toleran que un científico recorra la tabla de ventas con un GROUP BY de tres años. El ETL es el puente: lee sin escribir en origen, trabaja en un recinto propio y publica una foto.
1.1 ETL frente a ELT
En ETL se transforma antes de persistir el destino analítico: el motor es Python, Spark o un staging. En ELT se carga el crudo al almacén y la transformación es SQL dentro del destino. ELT no es “más moderno”: es dónde corre \(T\). Si el almacén es barato y el equipo piensa en SQL, ELT encaja. Si la lógica es un contrato que también usará el servicio online (sesión 9), conviene que \(T\) viva en código compartido —y eso empuja a ETL o a un híbrido.
1.2 OLTP, OLAP y el modelo
OLTP optimiza muchas transacciones cortas. OLAP (almacén, lakehouse) optimiza lecturas y agregados. El modelo de ML no debe leer OLTP “porque está fresco”: debe leer la foto que el ETL publicó. Mezclar las tres capas produce bloqueos en la base operativa y experimentos irreproducibles.
1.3 Qué transforma esta sesión — y qué no
Sí (estructural)
No (otra sesión)
Elegir columnas del contrato
Imputar nulos (3)
to_datetime, huso UTC
IQR / Isolation Forest (3)
Unir ventas con clientes por llave
One-hot, escalar (4)
total = precio * cantidad (grano de línea)
\(\log\), Yeo–Johnson (4)
Filtrar filas inválidas de esquema (precio no numérico)
Decidir MCAR/MAR (3)
from pathlib import Pathimport hashlibimport jsonimport sqlite3from datetime import datetime, timezoneimport numpy as npimport pandas as pdRUTA = Path("artefactos_sesion2")for sub in ("bronce", "plata", "oro"): (RUTA / sub).mkdir(parents=True, exist_ok=True)def hash_bytes(b: bytes) ->str:return hashlib.sha256(b).hexdigest()[:16]def hash_df(df: pd.DataFrame) ->str:return hash_bytes(df.to_csv(index=False).encode("utf-8"))print("capas:", [p.name for p insorted(RUTA.iterdir())])
capas: ['bronce', 'oro', 'plata']
2) Fuentes: el origen no se edita
Una fuente es un sistema con dueño ajeno. El extractor es un invitado. Tres reglas:
Lectura. Nunca UPDATE en origen “para dejarlo bonito”.
Copia bronce. Lo extraído se guarda tal cual, con marca de extracción.
Contrato. Columnas y tipos esperados se declaran antes de transformar.
Taxonomía mínima: archivo plano, libro Excel, tabla SQL, contrato JSON (API). El stream continuo se nombra y se deja para un curso de ingeniería de eventos.
2.1 Semilla del caso: ventas y clientes
Un comercio pequeño publica cada noche un CSV de líneas de venta y un JSON de clientes. Eso es el “origen”. El ETL no es el comercio: es quien copia y publica la foto.
Extracción completa. Se lee todo el origen. Simple, cara, y a veces la única opción (un CSV que se pisa cada noche).
Extracción incremental. Se lee lo que supera una marca de agua\(w\) (un timestamp o un entero monótono). Barata, y frágil si llegan filas tardías con \(t < w\).
El contrato de extracción declara: columnas, tipos, huso, y la marca \(w\) usada. Sin eso, “corrí el ETL” no es reproducible.
Un CSV “que se ve bien en Excel” puede ser ; + latin-1. Pandas, por defecto, asume , + UTF-8. El síntoma es un UnicodeDecodeError o una sola columna. El contrato debe fijar encoding y separador; no “probar hasta que abra”.
malo = RUTA /"bronce"/"ventas_latin1.csv"ventas_src.to_csv(malo, index=False, encoding="latin-1", sep=";")try: pd.read_csv(malo)print("abrió con defaults: peligroso")exceptExceptionas e:print("defaults fallan:", type(e).__name__)ok = pd.read_csv(malo, encoding="latin-1", sep=";")print("con contrato n =", len(ok), "| columnas =", list(ok.columns))
defaults fallan: ParserError
con contrato n = 16 | columnas = ['venta_id', 'cliente_id', 'ts_venta', 'precio', 'cantidad', 'canal']
3.3 Excel como origen (no como almacén)
xlsx = RUTA /"bronce"/"ventas_origen.xlsx"ventas_src.to_excel(xlsx, index=False)desde_excel = pd.read_excel(xlsx)print("n excel =", len(desde_excel), "| columnas =", list(desde_excel.columns))print("el contrato nombra la hoja; no se 'adivina' la segunda")
n excel = 16 | columnas = ['venta_id', 'cliente_id', 'ts_venta', 'precio', 'cantidad', 'canal']
el contrato nombra la hoja; no se 'adivina' la segunda
El Excel del origen es un archivo que alguien más edita. No es el data warehouse. Cargar hacia Excel como destino analítico es, en este curso, un anti-patrón: no hay tipos estables ni hash limpio.
3.4 Contrato JSON (la “API” sin red)
Una API REST es, para el ETL, un JSON con esquema. Depender de la red en un cuaderno didáctico enseña el fallo equivocado (timeout). Aquí se extrae de un archivo que simula la respuesta: el contrato es el mismo.
3.5 Tabla SQL: el origen se consulta, no se rediseña
conn_oltp = sqlite3.connect(RUTA /"bronce"/"oltp.db")ventas_src.assign(precio=ventas_src["precio"].astype(str)).to_sql("ventas", conn_oltp, if_exists="replace", index=False)extraido_sql = pd.read_sql("SELECT * FROM ventas", conn_oltp)conn_oltp.close()print(extraido_sql.head(3).to_string(index=False))print("el origen sigue intacto: no hay UPDATE")
venta_id cliente_id ts_venta precio cantidad canal
v000 c99 2026-09-10 08:00:00 20.0 4 tienda
v001 c02 2026-09-10 11:00:00 12.0 4 app
v002 c01 2026-09-10 14:00:00 12.0 4 tienda
el origen sigue intacto: no hay UPDATE
3.6 Marca de agua: incremental
Sea \(w\) el máximo ts_venta ya cargado. La siguiente corrida pide \(t > w\). Si una venta de las 07:00 llega a las 11:00, ya no entra. Eso se llama dato tardío y se trata más abajo; no se “arregla” bajando \(w\) a ojo en cada incidente.
ventas_bronce["ts_venta"] = pd.to_datetime(ventas_bronce["ts_venta"])w = pd.Timestamp("2026-09-11 08:00")inc = ventas_bronce.loc[ventas_bronce["ts_venta"] > w].copy()print("marca de agua w =", w)print("completas:", len(ventas_bronce), "| incrementales:", len(inc))print(inc[["venta_id", "ts_venta"]].to_string(index=False))
marca de agua w = 2026-09-11 08:00:00
completas: 16 | incrementales: 7
venta_id ts_venta
v009 2026-09-11 11:00:00
v010 2026-09-11 14:00:00
v011 2026-09-11 17:00:00
v012 2026-09-11 20:00:00
v013 2026-09-11 23:00:00
v014 2026-09-12 02:00:00
v015 2026-09-12 05:00:00
3.7 Completa frente a incremental: no es una preferencia estética
Criterio
Completa
Incremental
Costo
crece con el histórico
crece con el delta
Reprocesar un martes mal cargado
trivial (se vuelve a bajar todo)
exige backfill explícito
Datos tardíos
entran si siguen en origen
se pierden si \(t \le w\)
Idempotencia
overwrite del destino
merge por llave
Una regla prudente: completa mientras el origen quepa; incremental cuando el costo lo obligue, con una ventana de relectura (p. ej. \(w\) menos 48 h) para tardíos.
w cruda : 2026-09-11 08:00:00 → 7 filas
w - 48 h : 2026-09-09 08:00:00 → 16 filas (cubre tardíos recientes)
3.8 El esquema se valida al extraer
Si falta cliente_id, el fallo es de extracción, no de “el modelo no predice”. Validar al final es llegar tarde.
CONTRATO_VENTAS = ["venta_id", "cliente_id", "ts_venta", "precio", "cantidad", "canal"]def validar_esquema(df: pd.DataFrame, contrato: list[str], nombre: str) ->list[str]:return [f"{nombre}: falta {c}"for c in contrato if c notin df.columns]print(validar_esquema(ventas_bronce, CONTRATO_VENTAS, "ventas") or"ventas: ok")huérfano = ventas_bronce.drop(columns=["cliente_id"])print(validar_esquema(huérfano, CONTRATO_VENTAS, "ventas"))
ventas: ok
['ventas: falta cliente_id']
4) Transformar: estructura, no doctrina de nulos
La \(T\) de esta sesión es estructural: tipos, husos, categorías como texto normalizado de llave (no encoding de modelo), grano, totales de línea. No se imputa. No se recortan outliers. Un nulo que sobrevive se reporta y viaja a la sesión 3.
4.1 Selección de columnas del contrato
# El origen podría traer columnas de más (notas, dumps). El contrato recorta.ventas_t = ventas_bronce.loc[:, CONTRATO_VENTAS].copy()print(list(ventas_t.columns))
4.2 Tipos: el precio como número es un contrato, no un astype optimista
En el origen, precio mezcla 12.0 y "12,5". Un astype(float) revienta o, con coerción silenciosa, inventa nulos. Se documenta la regla: coma decimal → punto, luego numérico; lo que no convierta se marca y no se imputa.
def precio_a_float(s: pd.Series) -> pd.Series: txt = s.astype(str).str.replace(",", ".", regex=False)return pd.to_numeric(txt, errors="coerce")ventas_t["precio"] = precio_a_float(ventas_t["precio"])ventas_t["cantidad"] = pd.to_numeric(ventas_t["cantidad"], errors="coerce")print(ventas_t[["venta_id", "precio", "cantidad"]].head(6).to_string(index=False))print("precios no convertibles:", int(ventas_t["precio"].isna().sum()), "← viajan a la sesión 3")
4.3 Filtro de invalidez de esquema, no de calidad estadística
Un cantidad <= 0 no es un outlier: es una fila que no es una venta. Sacarla es hacer cumplir el grano. Un monto alto sí puede ser venta: eso lo juzga la sesión 3.
total = precio * cantidad no es ingeniería de características: es el importe de la transacción, una columna que el negocio ya nombra. Quien la calcule distinto en el dashboard y en el modelo tiene dos granos.
antes : ['APP', 'Web', 'app', 'tienda']
después: ['app', 'tienda', 'web']
4.6 Husos: todo timestamp se declara
ts_venta sin huso es una hora local de alguien. El contrato de este curso: se interpreta como America/Bogota y se publica en UTC. La sesión 1 exigía \(t_{\text{evento}}\); aquí se le pone huso.
4.7 Datos tardíos: el reloj del hecho no es el reloj de llegada
Una venta de las 07:00 que aparece a las 11:00 tiene \(t_{\text{hecho}} < w\) y \(t_{\text{llegada}} > w\). Si el incremental solo mira \(t_{\text{hecho}}\), se pierde. El arreglo limpio es persistir \(t_{\text{ingesta}}\) en bronce y releer una ventana; no “correr el ETL otra vez a mano”.
t_ingesta = pd.Timestamp("2026-09-12 11:00", tz="America/Bogota")ventas_t["ts_ingesta"] = t_ingestatardias = ventas_t.loc[ventas_t["ts_venta"] < t_ingesta - pd.Timedelta(hours=24)]print("ingesta:", t_ingesta)print("hechos con más de 24 h al llegar:", len(tardias))
ingesta: 2026-09-12 11:00:00-05:00
hechos con más de 24 h al llegar: 9
5) Integrar: el grano se declara antes del merge
Concatenar apila el mismo grano (dos días de ventas). Un mergecambia el grano si la llave no es única en ambos lados. Un cliente con dos filas convierte cada venta en dos: el total se duplica y el modelo (sesión 6) aprenderá un mundo que no existe.
print("unicidad cliente_id en dimensión:", clientes_bronce["cliente_id"].is_unique)m = ventas_t.merge(clientes_bronce, on="cliente_id", how="left")print("ventas", len(ventas_t), "→ join", len(m), "| si crece, el grano explotó")print("ventas sin cliente (c99):", int(m["ciudad"].isna().sum()))
unicidad cliente_id en dimensión: True
ventas 16 → join 16 | si crece, el grano explotó
ventas sin cliente (c99): 5
Un c99 sin dimensión no se “rellena” aquí. Se cuenta y se deja. Imputar ciudad es sesión 3. Borrar las ventas huérfanas es una decisión de negocio, no un default de pandas.
Una foto diaria en plata/fecha=2026-09-11/part.csv permite rehacer un día sin pisar el resto. No es un lakehouse: es una convención de carpetas.
m["fecha"] = m["ts_venta_utc"].dt.tz_convert("UTC").dt.date.astype(str)for fecha, parte in m.groupby("fecha"): carpeta = RUTA /"plata"/f"fecha={fecha}" carpeta.mkdir(exist_ok=True) parte.drop(columns=["fecha"]).to_csv(carpeta /"part.csv", index=False)print("particiones:", sorted(p.name for p in (RUTA /"plata").iterdir() if p.is_dir()))
base 10 + lote 8 → tras dos merges: 16 | únicos: 16
7) Gobernanza mínima: hash y linaje
Sin hash, “es el mismo CSV” es una opinión. El hash de esta sesión cubre la foto publicada, no el registro de experimentos (sesión 8). El linaje responde: de qué archivos, con qué \(w\), a qué hora, con qué código de reglas.
Dos CSV con las mismas filas en distinto orden tienen distinto hash si se hashea el archivo. Para comparar contenido se ordena por llave antes de hashear. El contrato debe decir cuál de los dos hashes se publica.
a = hash_df(m)b = hash_df(m.sample(frac=1, random_state=0))c = hash_df(m.sort_values("venta_id"))d = hash_df(m.sample(frac=1, random_state=1).sort_values("venta_id"))print("sin ordenar, permutado :", a != b)print("ordenado por llave :", c == d)
sin ordenar, permutado : True
ordenado por llave : True
8) ELT mínimo: la \(T\) en SQL
Mismo destino, la transformación corre después de cargar el bronce. El resultado debe poder reconciliarse con el ETL de Python. Si no se reconcilia, hay dos doctrinas.
conn = sqlite3.connect(RUTA /"plata"/"elt.db")ventas_bronce.to_sql("bronce_ventas", conn, if_exists="replace", index=False)q = ("SELECT venta_id, cliente_id, ""CAST(REPLACE(CAST(precio AS TEXT), ',', '.') AS FLOAT) AS precio, ""cantidad, LOWER(TRIM(canal)) AS canal, ""CAST(REPLACE(CAST(precio AS TEXT), ',', '.') AS FLOAT) * cantidad AS total ""FROM bronce_ventas WHERE cantidad > 0")plata_sql = pd.read_sql(q, conn)conn.close()print(plata_sql.head(3).to_string(index=False))print("n SQL =", len(plata_sql), "| n Python (ventas_t) =", len(ventas_t))
Editar el Excel de origen “para que pandas no falle”.
Llamar ETL a un dropna() seguido de un modelo.
Append del mismo lote y declarar después una “crisis de duplicados”.
Incremental sin ventana para tardíos.
merge sin comprobar unicidad de la dimensión.
Publicar plata sin hash ni \(w\).
Usar Excel como data warehouse.
11) Checklist
¿El origen quedó intacto?
¿Hay bronce y plata, no un único CSV “ya limpio”?
¿El contrato de columnas se validó al extraer?
¿Los nulos se cuentan y no se imputan aquí?
¿La carga es idempotente (overwrite o merge por llave)?
¿El linaje nombra \(w\), hashes y reglas?
12) Conclusiones
ETL es el contrato \(E\) de la sesión 1: mover \(x\) sin inventar \(f\). Esta sesión fijó fuentes, marca de agua, esquema, grano, modos de carga y linaje. La sesión 3 toma la plata y pregunta si merece entrenarse. Quien impute o encodee en este cuaderno está resolviendo el problema equivocado.
13) Taller
A. Cambie el encoding del CSV de origen y deje el extractor con defaults. Documente el error. Luego fije el contrato.
B. Fuerce un merge con dimensión duplicada (c01 dos veces). ¿Cuánto crece \(n\)? ¿Qué pasa con sum(total)?
C. Corra etl_noche dos veces con append al CSV de oro (modifique la función). ¿Qué invariante rompió?
D. Proponga una \(w\) y una ventana de relectura para un origen que entrega tardíos de hasta 36 h.
E. Explique en cinco líneas por qué total = precio * cantidad no es la \(T(x)\) de la sesión 4.
print("Inventario sesión 2:")for p insorted(RUTA.rglob("*")):if p.is_file():print(f" {p.relative_to(RUTA).as_posix():40s}{p.stat().st_size:6d} B")print("Taller A–E. Si imputó un nulo, vuelva a la sección 1.3.")
Inventario sesión 2:
bronce/clientes_origen.json 242 B
bronce/oltp.db 8192 B
bronce/ventas_latin1.csv 722 B
bronce/ventas_origen.csv 724 B
bronce/ventas_origen.xlsx 5597 B
oro/foto_noche.csv 1006 B
oro/foto_noche.json 77 B
oro/lineas_listas.csv 869 B
plata/dw.db 8192 B
plata/elt.db 8192 B
plata/fecha=2026-09-10/part.csv 575 B
plata/fecha=2026-09-11/part.csv 1024 B
plata/fecha=2026-09-12/part.csv 555 B
plata/linaje.json 387 B
plata/ventas.csv 1958 B
plata/ventas_append_doble.csv 3818 B
plata/ventas_clientes.csv 1958 B
Taller A–E. Si imputó un nulo, vuelva a la sección 1.3.
💬 ¿Te sirvió?
Deja en los comentarios una duda o un caso donde aplicarías esto — respondo todos. Sígueme para no perderte el próximo artículo de la serie y comparte con alguien que esté aprendiendo arquitectura de IA.
👉 El código completo está disponible para ejecutar directamente.