Las tres de la mañana y nadie mirando

Tienes el almacén del capítulo 15 y la carga incremental del capítulo 16. Falta la parte que nadie enseña y que es la que suena el teléfono: qué pasa cuando eso corre solo a las tres de la mañana y algo sale mal 🌙

Tres preguntas, y las tres tienen que tener respuesta antes de programar nada:

• ¿Alguien se entera de que falló, o se descubre el lunes en una reunión?
• ¿Qué quedó a medias? ¿El tablero está viejo, vacío o mentiroso?
• ¿Cómo se vuelve a correr sin romper lo que sí funcionó?

Un almacén para trastear

import os
import shutil
import sqlite3
import tempfile

carpeta = tempfile.mkdtemp()
almacen = os.path.join(carpeta, 'almacen.db')
shutil.copy('tienda.db', almacen)

con = sqlite3.connect(almacen)
con.execute("""CREATE TABLE carga_log (
    paso TEXT, corrio_en TEXT, filas INTEGER, estado TEXT, detalle TEXT)""")
con.commit()
print('almacen listo, con', con.execute('SELECT count(*) FROM pedidos').fetchone()[0], 'pedidos')

almacen listo, con 900 pedidos

El orquestador entero, en dieciocho líneas

Lo escribo a mano porque cuando veas Airflow quiero que reconozcas lo que hace por dentro. Un paso es un nombre, de qué otros pasos depende, y una función:

def anota(paso, filas, estado, detalle=''):
    con.execute('INSERT INTO carga_log VALUES (?, ?, ?, ?, ?)',
                (paso, '2026-08-21 03:00', filas, estado, detalle))
    con.commit()


def corre(pasos):
    hechos = set()
    for nombre, necesita, funcion in pasos:
        if not necesita.issubset(hechos):
            faltan = ', '.join(sorted(necesita - hechos))
            anota(nombre, 0, 'saltado', 'falta ' + faltan)
            print(f'{nombre:9} saltado, falta {faltan}')
            continue
        try:
            filas = funcion()
        except Exception as e:
            anota(nombre, 0, 'error', str(e))
            print(f'{nombre:9} ERROR   {e}')
            continue
        hechos.add(nombre)
        anota(nombre, filas, 'ok')
        print(f'{nombre:9} ok      {filas} filas')


print('el motor son 18 lineas')

el motor son 18 lineas

Fíjate en lo que hace y en lo que no hace. Si un paso falla, los que dependen de él no corren y quedan anotados como saltados. No se para todo de golpe ni se sigue como si nada: se sigue con lo que no dependía del que falló.

Eso es una de las palabras que vas a oír, el DAG. Es un dibujo de qué necesita qué, y la letra que importa es la A de acíclico: un paso no puede depender de sí mismo dando la vuelta, porque entonces no hay por dónde empezar.

Los cuatro pasos

def extrae():
    con.execute('DROP TABLE IF EXISTS cruda_pedidos')
    con.execute("""CREATE TABLE cruda_pedidos AS
                   SELECT *, '2026-08-21' AS cargado_en FROM pedidos""")
    return con.execute('SELECT count(*) FROM cruda_pedidos').fetchone()[0]


def limpia():
    con.execute('DROP TABLE IF EXISTS limpia_pedidos')
    con.execute("""CREATE TABLE limpia_pedidos AS
                   SELECT id, fecha, trim(canal) AS canal, monto, cargado_en
                   FROM cruda_pedidos
                   WHERE monto IS NOT NULL AND monto > 0""")
    return con.execute('SELECT count(*) FROM limpia_pedidos').fetchone()[0]


def revisa():
    entraron, quedaron = con.execute(
        'SELECT (SELECT count(*) FROM cruda_pedidos), '
        '       (SELECT count(*) FROM limpia_pedidos)').fetchone()
    perdidas = entraron - quedaron
    if perdidas > entraron * 0.05:
        raise ValueError(f'se cayeron {perdidas} de {entraron} pedidos al limpiar')
    return quedaron


def publica():
    con.execute('DROP TABLE IF EXISTS ventas_dw')
    con.execute("""CREATE TABLE ventas_dw AS
                   SELECT canal, substr(fecha, 1, 7) AS mes,
                          count(*) AS pedidos, round(sum(monto), 2) AS venta
                   FROM limpia_pedidos GROUP BY canal, mes""")
    return con.execute('SELECT count(*) FROM ventas_dw').fetchone()[0]


PASOS = [('extrae',  set(),      extrae),
         ('limpia',  {'extrae'}, limpia),
         ('revisa',  {'limpia'}, revisa),
         ('publica', {'revisa'}, publica)]

corre(PASOS)

extrae    ok      900 filas
limpia    ok      873 filas
revisa    ok      873 filas
publica   ok      72 filas

Cuatro en verde. Y mira revisa, que es el paso que casi nadie escribe: no transforma nada. Solo mira si el resultado del paso anterior tiene sentido, y si no lo tiene, revienta a propósito.

Se cayeron 27 pedidos al limpiar, de 900. Son los de monto nulo o cero, que ya estaban ahí. Menos del 5% que puse de tope, así que la carga sigue.

print(con.execute('SELECT round(sum(venta), 2) FROM ventas_dw').fetchone()[0],
      'soles en el tablero')

532653.85 soles en el tablero

Ahora que salga mal

Se rompe algo en el sistema de origen y una de cada siete ventas empieza a llegar con monto negativo. No es raro: pasa cuando alguien cambia cómo se guardan las devoluciones y no avisa.

con.execute('UPDATE pedidos SET monto = -1 WHERE id % 7 = 0')
con.commit()

corre(PASOS)

extrae    ok      900 filas
limpia    ok      747 filas
revisa    ERROR   se cayeron 153 de 900 pedidos al limpiar
publica   saltado, falta revisa

Ahí está el capítulo entero en cuatro líneas.

El paso de limpieza no falló: hizo su trabajo y devolvió 747 filas tan contento, porque descartar filas que no cumplen es literalmente lo que le pediste. Quien avisa es el chequeo, comparando contra lo que entró.

Y lo importante es lo que no pasó:

print(con.execute('SELECT round(sum(venta), 2) FROM ventas_dw').fetchone()[0],
      'soles en el tablero')

532653.85 soles en el tablero

El tablero sigue con los números de la carga buena. Está viejo, y está bien. Si publica hubiera corrido, la jefa de ventas abriría el lunes un tablero fresquito al que le faltan 153 pedidos, sin ninguna señal de que le falta nada 😐

Viejo y correcto se explica en una frase. Nuevo y a medias no se explica nunca, porque para explicarlo primero hay que darse cuenta, y nadie se da cuenta 💛

Un chequeo, visto de cerca

Llama al chequeo tú misma, sin el orquestador que lo atrapa:

revisa()

ValueError: se cayeron 153 de 900 pedidos al limpiar

Eso es todo lo que es un chequeo de calidad: una consulta y un raise. Las herramientas caras traen cientos escritos, y el que te va a salvar es el que escribas tú, porque conoce tu negocio.

Los cuatro que pondría siempre:

• Llegaron filas. Cero filas casi nunca es un día tranquilo.
• No se perdieron por el camino. El cuadre del capítulo 15, como paso.
• La clave no se repite. Un count(*) contra un count(DISTINCT id).
• Los números están en su rango. Ninguna venta negativa, ninguna fecha del futuro.

Y la bitácora, para el lunes

for fila in con.execute('SELECT paso, filas, estado, detalle FROM carga_log '
                        'ORDER BY rowid DESC LIMIT 4').fetchall():
    print('%-9s %4d  %-8s %s' % fila)

publica      0  saltado  falta revisa
revisa       0  error    se cayeron 153 de 900 pedidos al limpiar
limpia     747  ok
extrae     900  ok

Cuatro líneas que contestan qué pasó, en qué paso y por qué, sin que nadie tenga que acordarse. Es la misma bitácora del capítulo anterior con una columna más.

Las herramientas, para que no te vendan humo

Ninguna de las cuatro te va a decir qué chequeo poner ni cuál es el orden correcto. Eso lo pones tú, y por eso vale la pena entenderlo con dieciocho líneas antes de instalar nada 🧾

Ejercicios

import time

intentos = {'n': 0}


def baja_del_sistema():
    intentos['n'] += 1
    if intentos['n'] < 3:
        raise ConnectionError('timeout contra el servidor de pedidos')
    return 900


def con_reintentos(funcion, veces=4, espera=0.01):
    for intento in range(1, veces + 1):
        try:
            return funcion()
        except ConnectionError as e:
            print(f'intento {intento}: {e}')
            if intento == veces:
                raise
            time.sleep(espera * 2 ** (intento - 1))


print('filas:', con_reintentos(baja_del_sistema))

intento 1: timeout contra el servidor de pedidos
intento 2: timeout contra el servidor de pedidos
filas: 900

limpio = os.path.join(carpeta, 'limpio.db')
shutil.copy('tienda.db', limpio)
con = sqlite3.connect(limpio)


def carga_del_dia(dia):
    con.execute('DELETE FROM ventas_dia WHERE fecha = ?', (dia,))
    con.execute("""INSERT INTO ventas_dia
                   SELECT fecha, count(*), round(sum(monto), 2)
                   FROM pedidos WHERE fecha = ? GROUP BY fecha""", (dia,))
    con.commit()
    return con.execute('SELECT count(*) FROM ventas_dia WHERE fecha = ?',
                       (dia,)).fetchone()[0]


con.execute('CREATE TABLE ventas_dia (fecha TEXT, pedidos INTEGER, venta REAL)')

for dia in ('2026-06-20', '2026-06-21', '2026-06-22', '2026-06-23'):
    carga_del_dia(dia)

for fila in con.execute('SELECT * FROM ventas_dia ORDER BY fecha').fetchall():
    print(fila)

('2026-06-20', 2, 1224.52)
('2026-06-21', 2, 1450.95)
('2026-06-22', 2, 1806.89)
('2026-06-23', 2, 1541.36)

for dia in ('2026-06-20', '2026-06-21'):
    carga_del_dia(dia)

print(con.execute('SELECT count(*), round(sum(venta), 2) FROM ventas_dia').fetchone())

(4, 6023.72)

con.execute('CREATE TABLE saldo_canal (canal TEXT PRIMARY KEY, acumulado REAL)')
con.execute("INSERT INTO saldo_canal SELECT canal, 0 FROM pedidos GROUP BY canal")


def suma_el_dia(dia):
    con.execute("""UPDATE saldo_canal SET acumulado = acumulado + coalesce(
                       (SELECT sum(monto) FROM pedidos p
                        WHERE p.canal = saldo_canal.canal AND p.fecha = ?), 0)""",
                (dia,))
    con.commit()


suma_el_dia('2026-06-20')
print('una vez :', con.execute('SELECT round(sum(acumulado), 2) FROM saldo_canal').fetchone()[0])
suma_el_dia('2026-06-20')
print('dos veces:', con.execute('SELECT round(sum(acumulado), 2) FROM saldo_canal').fetchone()[0])

una vez : 1224.52
dos veces: 2449.04

Preguntas frecuentes