2  Fundamentos de ETL para datos de producción

mlops
etl
ingenieria-datos
Autor/a

Wilder Ramírez Delgado

Fecha de publicación

12 de septiembre de 2026

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:

  1. 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.
  2. 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.
  3. 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:

  1. Definir ETL como contrato de movimiento y distinguirlo de ELT, de la calidad y de \(T(x)\).
  2. Extraer desde archivo, tabla y “API” (contrato JSON) sin alterar el origen.
  3. Justificar extracción completa frente a incremental (marca de agua) y detectar datos tardíos.
  4. Aplicar transformaciones estructurales: tipos, husos, grano, uniones —sin imputar ni encodear para el modelo.
  5. Cargar en modos overwrite / append / merge, con capas bronce y plata, de forma idempotente.
  6. 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 Path
import hashlib
import json
import sqlite3
from datetime import datetime, timezone

import numpy as np
import pandas as pd

RUTA = 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 in sorted(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:

  1. Lectura. Nunca UPDATE en origen “para dejarlo bonito”.
  2. Copia bronce. Lo extraído se guarda tal cual, con marca de extracción.
  3. 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.

rng = np.random.default_rng(2)
n = 16
ventas_src = pd.DataFrame({
    "venta_id": [f"v{i:03d}" for i in range(n)],
    "cliente_id": rng.choice(["c01", "c02", "c03", "c04", "c99"], size=n),
    "ts_venta": pd.date_range("2026-09-10 08:00", periods=n, freq="3h"),
    "precio": rng.choice([9.5, 12.0, 20.0, 35.0, "12,5"], size=n),
    "cantidad": rng.integers(1, 5, size=n),
    "canal": rng.choice(["APP", "app", "Web", "tienda"], size=n),
})
clientes_src = pd.DataFrame({
    "cliente_id": ["c01", "c02", "c03", "c04"],
    "ciudad": ["Bogotá", "Medellín", "Cali", "Barranquilla"],
    "segmento": ["retail", "retail", "vip", "retail"],
})
ventas_src.to_csv(RUTA / "bronce" / "ventas_origen.csv", index=False)
clientes_src.to_json(RUTA / "bronce" / "clientes_origen.json", orient="records", force_ascii=False)
print(ventas_src.head(4).to_string(index=False))
print("clientes:", list(clientes_src["cliente_id"]))
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
    v003        c02 2026-09-10 17:00:00   35.0         3 tienda
clientes: ['c01', 'c02', 'c03', 'c04']

3) Extraer: completa, incremental y con contrato

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.

3.1 Lectura desde CSV

ventas_bronce = pd.read_csv(RUTA / "bronce" / "ventas_origen.csv")
print(ventas_bronce.dtypes)
print("n =", len(ventas_bronce), "| hash =", hash_df(ventas_bronce))
venta_id      object
cliente_id    object
ts_venta      object
precio        object
cantidad       int64
canal         object
dtype: object
n = 16 | hash = d0400773d232ce10

3.2 Encoding y separador no son detalles

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")
except Exception as 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.

clientes_bronce = pd.read_json(RUTA / "bronce" / "clientes_origen.json")
print(clientes_bronce.to_string(index=False))
print("hash clientes =", hash_df(clientes_bronce))
cliente_id       ciudad segmento
       c01       Bogotá   retail
       c02     Medellín   retail
       c03         Cali      vip
       c04 Barranquilla   retail
hash clientes = 4bc6eb78d8a99d81

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.

ventana = w - pd.Timedelta(hours=48)
relectura = ventas_bronce.loc[ventas_bronce["ts_venta"] > ventana]
print("w cruda      :", w, "→", len(inc), "filas")
print("w - 48 h     :", ventana, "→", len(relectura), "filas (cubre tardíos recientes)")
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 not in 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))
['venta_id', 'cliente_id', 'ts_venta', 'precio', 'cantidad', 'canal']

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")
venta_id  precio  cantidad
    v000    20.0         4
    v001    12.0         4
    v002    12.0         4
    v003    35.0         3
    v004    12.0         2
    v005    20.0         2
precios no convertibles: 0 ← 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.

antes = len(ventas_t)
ventas_t = ventas_t.loc[ventas_t["cantidad"] > 0].copy()
print(f"filas {antes} → {len(ventas_t)} (cantidad > 0)")
filas 16 → 16 (cantidad > 0)

4.4 Variable de grano: total de línea

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.

ventas_t["total"] = ventas_t["precio"] * ventas_t["cantidad"]
print(ventas_t[["venta_id", "precio", "cantidad", "total"]].head(4).to_string(index=False))
venta_id  precio  cantidad  total
    v000    20.0         4   80.0
    v001    12.0         4   48.0
    v002    12.0         4   48.0
    v003    35.0         3  105.0

4.5 Categorías como llaves, no como features

APP y app son el mismo canal. Unificar mayúsculas es consistencia de llave para un futuro JOIN o un groupby. No es one-hot. El one-hot es sesión 4.

print("antes :", sorted(ventas_t["canal"].unique()))
ventas_t["canal"] = ventas_t["canal"].str.strip().str.lower()
print("después:", sorted(ventas_t["canal"].unique()))
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.

ventas_t["ts_venta"] = pd.to_datetime(ventas_t["ts_venta"])
ventas_t["ts_venta"] = ventas_t["ts_venta"].dt.tz_localize(
    "America/Bogota", nonexistent="shift_forward", ambiguous="NaT"
)
ventas_t["ts_venta_utc"] = ventas_t["ts_venta"].dt.tz_convert("UTC")
print(ventas_t[["venta_id", "ts_venta", "ts_venta_utc"]].head(3).to_string(index=False))
venta_id                  ts_venta              ts_venta_utc
    v000 2026-09-10 08:00:00-05:00 2026-09-10 13:00:00+00:00
    v001 2026-09-10 11:00:00-05:00 2026-09-10 16:00:00+00:00
    v002 2026-09-10 14:00:00-05:00 2026-09-10 19:00:00+00:00

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_ingesta
tardias = 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 merge cambia 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.

5.1 Concatenación (mismo grano)

dia_a = ventas_t.head(5)
dia_b = ventas_t.tail(5)
apilado = pd.concat([dia_a, dia_b], ignore_index=True)
print(len(dia_a), "+", len(dia_b), "=", len(apilado))
5 + 5 = 10

5.2 merge y la cardinalidad

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.

print(m.loc[m["ciudad"].isna(), ["venta_id", "cliente_id", "total"]].head().to_string(index=False))
venta_id cliente_id  total
    v000        c99   80.0
    v005        c99   40.0
    v010        c99  105.0
    v012        c99   37.5
    v014        c99   48.0

6) Cargar: overwrite, append, merge

Modo Efecto Idempotente si…
overwrite reemplaza el destino se carga la misma foto
append agrega filas no se corre dos veces el mismo lote
merge upsert por llave la llave es única y estable

Append dos veces es el origen clásico de duplicados que la sesión 3 “descubrirá” como si fueran de captura.

destino = RUTA / "plata" / "ventas.csv"
m.to_csv(destino, index=False)  # overwrite
print("overwrite bytes:", destino.stat().st_size)
# append del mismo lote: error de operación, no de pandas
pd.concat([pd.read_csv(destino), m]).to_csv(RUTA / "plata" / "ventas_append_doble.csv", index=False)
print("append doble   :", len(pd.read_csv(RUTA / "plata" / "ventas_append_doble.csv")), "filas (duplicó)")
overwrite bytes: 1958
append doble   : 32 filas (duplicó)

6.1 Bronce, plata, oro

  • Bronce: copia del origen, sin doctrina.
  • Plata: esquema del contrato, tipos, uniones, sin imputar.
  • Oro: recortes listos para un consumidor (un dashboard o, más adelante, el train). No es “el feature store”.

El modelo de la sesión 6 debería leer plata u oro, nunca bronce “porque tiene más columnas”.

m.to_csv(RUTA / "plata" / "ventas_clientes.csv", index=False)
oro = m.loc[:, ["venta_id", "cliente_id", "ts_venta_utc", "total", "canal", "segmento"]]
oro.to_csv(RUTA / "oro" / "lineas_listas.csv", index=False)
print("bronce:", list((RUTA / "bronce").iterdir()))
print("plata :", list((RUTA / "plata").iterdir()))
print("oro   :", list((RUTA / "oro").iterdir()))
bronce: [WindowsPath('artefactos_sesion2/bronce/clientes_origen.json'), WindowsPath('artefactos_sesion2/bronce/oltp.db'), WindowsPath('artefactos_sesion2/bronce/ventas_latin1.csv'), WindowsPath('artefactos_sesion2/bronce/ventas_origen.csv'), WindowsPath('artefactos_sesion2/bronce/ventas_origen.xlsx')]
plata : [WindowsPath('artefactos_sesion2/plata/ventas.csv'), WindowsPath('artefactos_sesion2/plata/ventas_append_doble.csv'), WindowsPath('artefactos_sesion2/plata/ventas_clientes.csv')]
oro   : [WindowsPath('artefactos_sesion2/oro/lineas_listas.csv')]

6.2 Carga particionada por fecha

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()))
particiones: ['fecha=2026-09-10', 'fecha=2026-09-11', 'fecha=2026-09-12']

6.3 Destino SQL analítico (no el OLTP)

conn_dw = sqlite3.connect(RUTA / "plata" / "dw.db")
m.drop(columns=["fecha"], errors="ignore").to_sql("lineas", conn_dw, if_exists="replace", index=False)
n_dw = pd.read_sql("SELECT COUNT(*) AS n FROM lineas", conn_dw).iloc[0, 0]
conn_dw.close()
print("filas en dw.lineas =", n_dw)
filas en dw.lineas = 16

6.4 Idempotencia: correr dos veces no inventa hechos

Overwrite de la misma foto es idempotente. Append no. Merge por venta_id sí, si la llave es única.

def cargar_merge(existente: pd.DataFrame, lote: pd.DataFrame, llave: str) -> pd.DataFrame:
    resto = existente.loc[~existente[llave].isin(lote[llave])]
    return pd.concat([resto, lote], ignore_index=True)


base = m.head(10)
lote2 = m.tail(8)
dos_veces = cargar_merge(cargar_merge(base, lote2, "venta_id"), lote2, "venta_id")
print("base", len(base), "+ lote", len(lote2), "→ tras dos merges:", len(dos_veces),
      "| únicos:", dos_veces["venta_id"].nunique())
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.

linaje = {
    "t_ingesta": str(t_ingesta),
    "marca_agua_usada": str(w),
    "hash_bronce_ventas": hash_df(ventas_bronce),
    "hash_bronce_clientes": hash_df(clientes_bronce),
    "hash_plata": hash_df(m.drop(columns=["fecha"], errors="ignore")),
    "n_plata": int(len(m)),
    "n_sin_dimension": int(m["ciudad"].isna().sum()),
    "reglas": ["precio_coma_a_punto", "cantidad>0", "canal.lower", "huso=America/Bogota→UTC"],
}
(RUTA / "plata" / "linaje.json").write_text(
    json.dumps(linaje, indent=2, ensure_ascii=False), encoding="utf-8"
)
print(json.dumps(linaje, indent=2, ensure_ascii=False))
{
  "t_ingesta": "2026-09-12 11:00:00-05:00",
  "marca_agua_usada": "2026-09-11 08:00:00",
  "hash_bronce_ventas": "d0400773d232ce10",
  "hash_bronce_clientes": "4bc6eb78d8a99d81",
  "hash_plata": "561c9b29b1b8ebdb",
  "n_plata": 16,
  "n_sin_dimension": 5,
  "reglas": [
    "precio_coma_a_punto",
    "cantidad>0",
    "canal.lower",
    "huso=America/Bogota→UTC"
  ]
}

7.1 El hash cambia si cambia el orden

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))
venta_id cliente_id  precio  cantidad  canal  total
    v000        c99    20.0         4 tienda   80.0
    v001        c02    12.0         4    app   48.0
    v002        c01    12.0         4 tienda   48.0
n SQL = 16 | n Python (ventas_t) = 16

9) Pipeline integrado

Una sola función orquesta E→T→L y escribe linaje. No es Airflow. Es el guion que la sesión 7 convertirá en DAG.

def etl_noche(path_ventas: Path, path_clientes: Path, w: pd.Timestamp) -> dict:
    ventas = pd.read_csv(path_ventas)
    clientes = pd.read_json(path_clientes)
    faltan = validar_esquema(ventas, CONTRATO_VENTAS, "ventas")
    if faltan:
        raise ValueError(faltan)
    ventas["ts_venta"] = pd.to_datetime(ventas["ts_venta"])
    ventas = ventas.loc[ventas["ts_venta"] > w].copy()
    ventas["precio"] = precio_a_float(ventas["precio"])
    ventas["cantidad"] = pd.to_numeric(ventas["cantidad"], errors="coerce")
    ventas = ventas.loc[ventas["cantidad"] > 0].copy()
    ventas["canal"] = ventas["canal"].str.strip().str.lower()
    ventas["total"] = ventas["precio"] * ventas["cantidad"]
    foto = ventas.merge(clientes, on="cliente_id", how="left")
    foto.to_csv(RUTA / "oro" / "foto_noche.csv", index=False)
    meta = {"w": str(w), "n": int(len(foto)), "hash": hash_df(foto)}
    (RUTA / "oro" / "foto_noche.json").write_text(json.dumps(meta, indent=2), encoding="utf-8")
    return meta


meta = etl_noche(
    RUTA / "bronce" / "ventas_origen.csv",
    RUTA / "bronce" / "clientes_origen.json",
    w=pd.Timestamp("2026-09-10 06:00"),
)
print(meta)
{'w': '2026-09-10 06:00:00', 'n': 16, 'hash': '5352d754b4abe151'}

10) Errores frecuentes

  1. Editar el Excel de origen “para que pandas no falle”.
  2. Llamar ETL a un dropna() seguido de un modelo.
  3. Append del mismo lote y declarar después una “crisis de duplicados”.
  4. Incremental sin ventana para tardíos.
  5. merge sin comprobar unicidad de la dimensión.
  6. Publicar plata sin hash ni \(w\).
  7. Usar Excel como data warehouse.

11) Checklist

  1. ¿El origen quedó intacto?
  2. ¿Hay bronce y plata, no un único CSV “ya limpio”?
  3. ¿El contrato de columnas se validó al extraer?
  4. ¿Los nulos se cuentan y no se imputan aquí?
  5. ¿La carga es idempotente (overwrite o merge por llave)?
  6. ¿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 in sorted(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.