{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Las tres de la mañana y nadie mirando\n", "\n", "Pasos que dependen unos de otros, chequeos que paran la carga antes de publicar, reintentos y bitácora. Con un orquestador de dieciocho líneas para entender qué hacen los grandes.\n", "\n", "Cuaderno de práctica del capítulo 17 de **SQL desde cero**, de Miss Yera.\n", "\n", "Corre de arriba abajo. Si lo abres en Google Colab no necesitas instalar nada.\n", "\n", "Capítulo completo: https://missyera.com/guias/sql-desde-cero/cuando-la-carga-falla/\n", "\n", "Los ejercicios están al final y traen una celda vacía debajo de cada uno. Las\n", "respuestas viven en el cuaderno de soluciones, y merece la pena pelearse un\n", "rato antes de abrirlo 💛" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Antes de empezar\n", "\n", "Se baja la base y se deja lista una función `q()` que corre las consultas." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import sqlite3\n", "import urllib.request\n", "\n", "import pandas as pd\n", "\n", "urllib.request.urlretrieve(\"https://missyera.com/static/datasets/tienda.db\", \"tienda.db\")\n", "con = sqlite3.connect(\"tienda.db\")\n", "\n", "def q(sql):\n", " \"\"\"Corre las sentencias del bloque y devuelve la ultima como tabla.\n", "\n", " Parte por sentencias igual que la consola de SQLite, porque un bloque puede\n", " traer varias y un CREATE TRIGGER lleva punto y coma dentro de su cuerpo.\n", " \"\"\"\n", " resultado = None\n", " trozo = \"\"\n", " for linea in sql.splitlines(keepends=True):\n", " trozo += linea\n", " if sqlite3.complete_statement(trozo):\n", " if trozo.strip():\n", " cur = con.execute(trozo.strip())\n", " resultado = (pd.DataFrame(cur.fetchall(),\n", " columns=[d[0] for d in cur.description])\n", " if cur.description else None)\n", " trozo = \"\"\n", " if trozo.strip():\n", " cur = con.execute(trozo.strip())\n", " resultado = (pd.DataFrame(cur.fetchall(),\n", " columns=[d[0] for d in cur.description])\n", " if cur.description else None)\n", " con.commit()\n", " return resultado\n", "\n", "q(\"SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY name\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Antes de empezar\n", "\n", "Esta celda baja el ayudante que corrige tus ejercicios. Después, en cada\n", "ejercicio que se pueda corregir solo, vas a ver `%%revisa` arriba de la celda:\n", "escribe tu respuesta debajo, ejecuta, y te digo si te salió 💛" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import urllib.request\n", "\n", "# El ayudante de los cuadernos. Trae la corrección de los ejercicios y, en los\n", "# capítulos de consola, la celda mágica que ejecuta los comandos. Se baja en\n", "# vez de venir pegado aquí para que siempre sea el último.\n", "urllib.request.urlretrieve(\n", " \"https://missyera.com/static/cuadernos/revisa.py\", \"revisa.py\")\n", "import revisa\n", "revisa.carga({\n", " 1: \"aW50ZW50byAxOiB0aW1lb3V0IGNvbnRyYSBlbCBzZXJ2aWRvciBkZSBwZWRpZG9zCmludGVudG8gMjogdGltZW91dCBjb250cmEgZWwgc2Vydmlkb3IgZGUgcGVkaWRvcwpmaWxhczogOTAw\",\n", " 2: \"KCcyMDI2LTA2LTIwJywgMiwgMTIyNC41MikKKCcyMDI2LTA2LTIxJywgMiwgMTQ1MC45NSkKKCcyMDI2LTA2LTIyJywgMiwgMTgwNi44OSkKKCcyMDI2LTA2LTIzJywgMiwgMTU0MS4zNik=\",\n", " 3: \"KDQsIDYwMjMuNzIp\",\n", " 4: \"dW5hIHZleiA6IDEyMjQuNTIKZG9zIHZlY2VzOiAyNDQ5LjA0\",\n", "}, lenguaje=\"sql\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Tienes el almacén del capítulo 15 y la carga\n", "incremental del capítulo 16. Falta la parte que nadie\n", "enseña y que es la que suena el teléfono: **qué pasa cuando eso corre solo\n", "a las tres de la mañana y algo sale mal** 🌙\n", "\n", "Tres preguntas, y las tres tienen que tener respuesta antes de programar\n", "nada:\n", "\n", "- ¿Alguien se entera de que falló, o se descubre el lunes en una reunión?\n", "\n", "- ¿Qué quedó a medias? ¿El tablero está viejo, vacío o mentiroso?\n", "\n", "- ¿Cómo se vuelve a correr sin romper lo que sí funcionó?" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Un almacén para trastear" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import os\n", "import shutil\n", "import sqlite3\n", "import tempfile\n", "\n", "carpeta = tempfile.mkdtemp()\n", "almacen = os.path.join(carpeta, 'almacen.db')\n", "shutil.copy('tienda.db', almacen)\n", "\n", "con = sqlite3.connect(almacen)\n", "con.execute(\"\"\"CREATE TABLE carga_log (\n", " paso TEXT, corrio_en TEXT, filas INTEGER, estado TEXT, detalle TEXT)\"\"\")\n", "con.commit()\n", "print('almacen listo, con', con.execute('SELECT count(*) FROM pedidos').fetchone()[0], 'pedidos')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## El orquestador entero, en dieciocho líneas" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Lo escribo a mano porque cuando veas Airflow quiero que reconozcas lo que\n", "hace por dentro. Un paso es un nombre, de qué otros pasos depende, y una función:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def anota(paso, filas, estado, detalle=''):\n", " con.execute('INSERT INTO carga_log VALUES (?, ?, ?, ?, ?)',\n", " (paso, '2026-08-21 03:00', filas, estado, detalle))\n", " con.commit()\n", "\n", "\n", "def corre(pasos):\n", " hechos = set()\n", " for nombre, necesita, funcion in pasos:\n", " if not necesita.issubset(hechos):\n", " faltan = ', '.join(sorted(necesita - hechos))\n", " anota(nombre, 0, 'saltado', 'falta ' + faltan)\n", " print(f'{nombre:9} saltado, falta {faltan}')\n", " continue\n", " try:\n", " filas = funcion()\n", " except Exception as e:\n", " anota(nombre, 0, 'error', str(e))\n", " print(f'{nombre:9} ERROR {e}')\n", " continue\n", " hechos.add(nombre)\n", " anota(nombre, filas, 'ok')\n", " print(f'{nombre:9} ok {filas} filas')\n", "\n", "\n", "print('el motor son 18 lineas')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Fíjate en lo que hace y en lo que no hace. Si un paso falla, los que dependen\n", "de él **no corren** y quedan anotados como saltados. No se para todo\n", "de golpe ni se sigue como si nada: se sigue con lo que no dependía del que\n", "falló.\n", "\n", "Eso es una de las palabras que vas a oír, el **DAG**. Es un\n", "dibujo de qué necesita qué, y la letra que importa es la A de acíclico: un paso\n", "no puede depender de sí mismo dando la vuelta, porque entonces no hay por dónde\n", "empezar." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Los cuatro pasos" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def extrae():\n", " con.execute('DROP TABLE IF EXISTS cruda_pedidos')\n", " con.execute(\"\"\"CREATE TABLE cruda_pedidos AS\n", " SELECT *, '2026-08-21' AS cargado_en FROM pedidos\"\"\")\n", " return con.execute('SELECT count(*) FROM cruda_pedidos').fetchone()[0]\n", "\n", "\n", "def limpia():\n", " con.execute('DROP TABLE IF EXISTS limpia_pedidos')\n", " con.execute(\"\"\"CREATE TABLE limpia_pedidos AS\n", " SELECT id, fecha, trim(canal) AS canal, monto, cargado_en\n", " FROM cruda_pedidos\n", " WHERE monto IS NOT NULL AND monto > 0\"\"\")\n", " return con.execute('SELECT count(*) FROM limpia_pedidos').fetchone()[0]\n", "\n", "\n", "def revisa():\n", " entraron, quedaron = con.execute(\n", " 'SELECT (SELECT count(*) FROM cruda_pedidos), '\n", " ' (SELECT count(*) FROM limpia_pedidos)').fetchone()\n", " perdidas = entraron - quedaron\n", " if perdidas > entraron * 0.05:\n", " raise ValueError(f'se cayeron {perdidas} de {entraron} pedidos al limpiar')\n", " return quedaron\n", "\n", "\n", "def publica():\n", " con.execute('DROP TABLE IF EXISTS ventas_dw')\n", " con.execute(\"\"\"CREATE TABLE ventas_dw AS\n", " SELECT canal, substr(fecha, 1, 7) AS mes,\n", " count(*) AS pedidos, round(sum(monto), 2) AS venta\n", " FROM limpia_pedidos GROUP BY canal, mes\"\"\")\n", " return con.execute('SELECT count(*) FROM ventas_dw').fetchone()[0]\n", "\n", "\n", "PASOS = [('extrae', set(), extrae),\n", " ('limpia', {'extrae'}, limpia),\n", " ('revisa', {'limpia'}, revisa),\n", " ('publica', {'revisa'}, publica)]\n", "\n", "corre(PASOS)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Cuatro en verde. Y mira `revisa`, que es el paso que casi nadie\n", "escribe: **no transforma nada**. Solo mira si el resultado del paso\n", "anterior tiene sentido, y si no lo tiene, revienta a propósito.\n", "\n", "Se cayeron 27 pedidos al limpiar, de 900. Son los de monto nulo o cero, que\n", "ya estaban ahí. Menos del 5% que puse de tope, así que la carga sigue." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print(con.execute('SELECT round(sum(venta), 2) FROM ventas_dw').fetchone()[0],\n", " 'soles en el tablero')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Ahora que salga mal" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Se rompe algo en el sistema de origen y una de cada siete ventas empieza a\n", "llegar con monto negativo. No es raro: pasa cuando alguien cambia cómo se\n", "guardan las devoluciones y no avisa." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "con.execute('UPDATE pedidos SET monto = -1 WHERE id % 7 = 0')\n", "con.commit()\n", "\n", "corre(PASOS)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Ahí está el capítulo entero en cuatro líneas.\n", "\n", "El paso de limpieza **no falló**: hizo su trabajo y devolvió 747\n", "filas tan contento, porque descartar filas que no cumplen es literalmente lo que\n", "le pediste. Quien avisa es el chequeo, comparando contra lo que entró.\n", "\n", "Y lo importante es lo que no pasó:" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print(con.execute('SELECT round(sum(venta), 2) FROM ventas_dw').fetchone()[0],\n", " 'soles en el tablero')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "El tablero sigue con los números de la carga buena. Está viejo, y está bien.\n", "Si `publica` hubiera corrido, la jefa de ventas abriría el lunes un\n", "tablero fresquito al que le faltan 153 pedidos, sin ninguna señal de que le\n", "falta nada 😐\n", "\n", "**Viejo y correcto se explica en una frase. Nuevo y a medias no se\n", "explica nunca**, porque para explicarlo primero hay que darse cuenta, y\n", "nadie se da cuenta 💛" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Un chequeo, visto de cerca" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Llama al chequeo tú misma, sin el orquestador que lo atrapa:" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Esto revienta a propósito.** Se ejecuta dentro de un `try` para que puedas seguir con \"ejecutar todo\" y aun así ver la queja." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "try:\n", " revisa()\n", "except Exception as e:\n", " print(f'{type(e).__name__}: {e}')" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Y la queja que tiene que salir es esta:\n", "\n", "```\n", "ValueError: se cayeron 153 de 900 pedidos al limpiar\n", "```" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Eso es todo lo que es un chequeo de calidad: una consulta y un\n", "`raise`. Las herramientas caras traen cientos escritos, y el que te\n", "va a salvar es el que escribas tú, porque conoce tu negocio.\n", "\n", "Los cuatro que pondría siempre:\n", "\n", "- **Llegaron filas.** Cero filas casi nunca es un día tranquilo.\n", "\n", "- **No se perdieron por el camino.** El cuadre del capítulo\n", "15, como paso.\n", "\n", "- **La clave no se repite.** Un `count(*)` contra un\n", "`count(DISTINCT id)`.\n", "\n", "- **Los números están en su rango.** Ninguna venta negativa,\n", "ninguna fecha del futuro." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Y la bitácora, para el lunes" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "for fila in con.execute('SELECT paso, filas, estado, detalle FROM carga_log '\n", " 'ORDER BY rowid DESC LIMIT 4').fetchall():\n", " print('%-9s %4d %-8s %s' % fila)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Cuatro líneas que contestan qué pasó, en qué paso y por qué, sin que nadie\n", "tenga que acordarse. Es la misma bitácora del capítulo anterior con una columna\n", "más." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Las herramientas, para que no te vendan humo" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "| Herramienta | Qué agrega sobre lo de arriba | Cuándo |\n", "|---|---|---|\n", "| cron | lo dispara a una hora, y nada más | una o dos cargas simples |\n", "| Airflow | el DAG con pantalla, reintentos, backfill, historial | muchas cargas que dependen entre sí |\n", "| Dagster, Prefect | lo mismo con otra filosofía y menos ceremonia | igual, si empiezas de cero hoy |\n", "| dbt | ordena y prueba las transformaciones en SQL, no las mueve | cuando la parte difícil ya es el SQL |" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Ninguna de las cuatro te va a decir qué chequeo poner ni cuál es el orden\n", "correcto. Eso lo pones tú, y por eso vale la pena entenderlo con dieciocho\n", "líneas antes de instalar nada 🧾" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Programar que una consulta corra sola cada noche\n", "\n", "| PostgreSQL | `no trae planificador: se usa la extension pg_cron o el cron del sistema` |\n", "|---|---|\n", "| MySQL | `CREATE EVENT ... ON SCHEDULE EVERY 1 DAY, si event_scheduler esta en ON` |\n", "| SQL Server | `SQL Server Agent, que es un servicio aparte y hay que tenerlo instalado` |\n", "| SQLite | `nada: no hay ningun proceso corriendo al que programarle algo` |\n", "\n", "Y de los cuatro, el que mas se usa en proyectos de verdad es ninguno: el planificador vive fuera de la base, porque una carga casi nunca es solo SQL. Baja un archivo, llama a una API, escribe en otro sitio. La base es un paso, no el director." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Comprueba que lo tienes\n", "\n", "La carga de anoche se cayó en el paso de limpieza y el paso que publica el tablero no llegó a correr. ¿Qué es lo mejor que puede pasar a la mañana siguiente?\n", "\n", "a) Que el tablero enseñe los números de anteayer y la bitácora diga que la carga falló\n", "\n", "b) Que el tablero salga vacío para que se note\n", "\n", "c) Que el tablero publique lo que alcanzó a limpiarse\n", "\n", "d) Que la carga se reintente sola hasta que pase" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Ejercicios" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 1. Reintentar lo que se puede reintentar\n", "\n", "El servidor de origen se cae a ratos. Escribe el reintento\n", "con espera creciente." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%%revisa 1\n", "-- tu turno" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 2. Volver a correr días viejos\n", "\n", "Escribe una carga por día que se pueda relanzar para\n", "cualquier fecha. Arriba le metimos montos negativos a la base a propósito, así\n", "que empiezo copiándola otra vez limpia." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%%revisa 2\n", "-- tu turno" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 3. Comprueba que el backfill no ensucia\n", "\n", "Relanza dos días que ya estaban cargados." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%%revisa 3\n", "-- tu turno" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 4. El paso que no se puede reintentar\n", "\n", "Un acumulador. Córrelo dos veces y mira qué pasa." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "%%revisa 4\n", "-- tu turno" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 5. Cuándo suena el teléfono\n", "\n", "Tienes la bitácora. ¿Qué merece despertar a alguien a las\n", "tres de la mañana y qué espera al desayuno?" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# tu turno" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 6. Ordena estos seis pasos\n", "\n", "Una carga real, desordenada. Di qué depende de qué." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# tu turno" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Preguntas frecuentes" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "¿Qué es un DAG?El dibujo de qué paso necesita a qué paso. La A es de acíclico: un paso no puede depender de sí mismo dando la vuelta, porque entonces no habría por dónde empezar.\n", "\n", "¿Qué es dbt?Una herramienta para ordenar y probar transformaciones escritas en SQL. No mueve datos: los transforma donde ya están, y su aporte real son las pruebas y la documentación.\n", "\n", "¿Qué es Airflow?Un orquestador: dispara los pasos en el orden del DAG, reintenta lo que falla y guarda el historial. Por dentro hace lo mismo que las dieciocho líneas de este capítulo, con pantalla.\n", "\n", "¿Qué es un backfill?Volver a correr la carga para fechas pasadas. Solo se puede si la carga recibe el día como parámetro, y por eso se escribe así desde el principio." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "---\n", "\n", "Ese era el capítulo 17 de **SQL desde cero**. El texto completo, con las salidas de cada bloque, está en https://missyera.com/guias/sql-desde-cero/cuando-la-carga-falla/\n", "\n", "Que tengas lindo día! 🌸" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python", "version": "3.11" } }, "nbformat": 4, "nbformat_minor": 5 }