Tubería ETL, pensé que tenía una idea bastante clara de lo que realmente era la ingeniería de datos. Extraes datos de algún lugar, los limpias, los cargas en algún lugar útil. ETL. Bastante simple.
Por contexto, soy un analista de datos que intenta hacer la transición a la ingeniería de datos. He estado documentando ese viaje públicamente, comenzando con una hoja de ruta de autoestudio de 12 meses que elaboré a principios de este año. El paso más reciente en ese viaje fue construir mi primer canal ETL desde cero usando la API de GitHub, sobre la cual escribí aquí en TDS. Ese oleoducto funcionó. Extrajo datos, los limpió y los guardó en un CSV. Estaba feliz con eso.
Así que decidí impulsarlo más allá, hacerlo más “listo para producción”, como le gusta decir a Internet. Lo que pasó después realmente me sorprendió. No porque las cosas se rompieran, sino por lo que la ruptura reveló.
El oleoducto original
La tubería original era básica, lo cual estaba bien porque ese era el punto. Extraiga datos de la API de GitHub, limpie un poco y guarde todo en un archivo CSV. Funcionó perfectamente para lo que era: un ejercicio de aprendizaje. Pero la ingeniería de datos no funciona en el mundo real con un archivo CSV y un script de un solo uso. Quería saber qué significaba realmente el “mundo real” en la práctica, así que decidí seguir adelante y ver qué sucedía.
Aquí está el canal original completo para cualquiera que no haya leído el artículo anterior:
importar solicitudes desde fecha y hora importar fecha y hora, timedelta url = “https://api.github.com/search/repositories” params = { “q”: “language:python creado:>2025-04-22”, “sort”: “stars”, “order”: “desc”, “per_page”: 30 } respuesta = request.get(url, params=params) data = respuesta.json() importar pandas como pd repositorios = []
para repositorio en datos[‘items’]: repos.append({ “nombre”: repositorio[‘name’]”propietario”: repositorio[‘owner’][‘login’]”estrellas”: repositorio[‘stargazers_count’]”bifurcaciones”: repositorio[‘forks_count’]”idioma”: repositorio[‘language’]”descripción”: repositorio[‘description’]”url”: repositorio[‘html_url’]”created_at”: repositorio[‘created_at’]
}) df = pd.DataFrame(repositorios) df_clean = df.dropna(subconjunto=[‘description’]) df_clean = df_clean.copia() df_clean[‘viral’] =df_clean[‘stars’].apply(lambda x: ‘Sí’ si x > 50000 else ‘No’) df_clean.sort_values(‘stars’, ascending=False).reset_index(drop=True) df_clean.to_csv(‘github_trending_repos.csv’, index=False) print(“Tubería completa. Archivo guardado.”)
Simple, legible y funciona. Pero en el momento en que intentas ejecutarlo más de una vez, o volver a usarlo al día siguiente, las grietas comienzan a aparecer.
Primer muro: El oleoducto no tenía memoria
La primera actualización fue sencilla. En lugar de guardarlos en un archivo CSV, cargué los datos en una base de datos SQLite. SQLite sigue siendo un solo archivo, pero se comporta como una base de datos real. Puede consultarlo, verificar lo que ya contiene y construir sobre él correctamente. Se sintió como un pequeño cambio. No lo fue.
Ejecuté el proceso una vez y obtuve 22 repositorios. Luego lo ejecuté por segunda vez sin cambiar nada y verifiqué la base de datos.
Filas totales: 44 Repositorios únicos: 22 Duplicados: 22
Sinceramente, no me lo esperaba. Supuse que podría suceder, pero nunca pensé que sucedería. Pero me alegro de que así fuera, porque fue la primera vez que vi cómo se rompía mi tubería. Y lo que reveló fue simple pero importante: el guión no tenía memoria. Cada vez que se ejecutaba, comenzaba de nuevo y agregaba ciegamente todo lo que encontraba. Sin advertencia, sin error. Simplemente “Pipeline complete” como si todo estuviera bien.
Aquí es donde me encontré con un concepto llamado idempotencia.
Idempotencia es una palabra elegante para una idea simple. Si algo ya pasó, no debería volver a pasar. En el contexto de una canalización de datos, significa que ejecutar su canalización una vez o diez veces siempre debería producir el mismo resultado. Sin filas adicionales, sin duplicados, sin corrupción silenciosa de sus datos.
En principio, la solución fue sencilla. Antes de insertar algo en la base de datos, la canalización ahora verifica si ese registro ya existe. Si es así, primero lo elimina y luego inserta la versión nueva. Un pequeño cambio en la forma de pensar, pero cambia por completo la confiabilidad de su canalización.
Y aquí está la parte que se me quedó grabada: un guión básico nunca pensará en esto por sí solo. Tienes que incorporarlo deliberadamente. Eso ya no es un guión. Eso es ingeniería.
Muro dos: los datos desaparecieron de la noche a la mañana
El segundo muro era menos técnico y más inquietante.
Cuando cerré Colab por la noche y regresé al día siguiente, sentí una sensación de inquietud. Tuve que ejecutar todo nuevamente desde cero y esperar que nada se rompiera, a pesar de que todo había funcionado perfectamente la noche anterior. La base de datos que había construido cuidadosamente ya no estaba. Y recordé que antes de este proyecto, había tenido dificultades para encontrar mi archivo de canalización ETL original. Pasé tiempo buscándolo hasta que finalmente lo hice. Esa sensación de casi perder el trabajo permanece contigo.
Sabía que tenía que haber una manera mejor. Un oleoducto real no puede depender de que haya alguien allí para volver a ejecutarlo todas las mañanas. Los datos tienen que vivir en algún lugar que sobreviva más allá de la sesión.
La solución aquí fue montar Google Drive directamente dentro de Colab y apuntar la conexión de la base de datos allí en lugar de al entorno temporal de Colab. Cambio de una línea:
conexión = sqlite3.connect(‘/content/drive/MyDrive/github_repos.db’)
Ahora la base de datos se encuentra en Google Drive. Cierre la sesión, reinicie el tiempo de ejecución, abra un cuaderno completamente nuevo. Los datos siguen ahí esperando.
Pero esta solución reveló algo más grande. Si la persistencia de los datos ya requiere pensar en dónde viven las cosas y cómo sobreviven, ¿qué sucede cuando necesitas que la canalización se ejecute automáticamente todos los días sin que tú la toques?
Tercer muro: nadie puede presionar Run Forever
La tercera pared fue la que más me emocionó, la que me sorprendió.
Incluso con el problema de duplicación resuelto y la base de datos viviendo de forma segura en Google Drive, alguien todavía tiene que abrir el cuaderno y presionar ejecutar. Colab no es un servidor. Es un entorno interactivo. No se despierta a las 3 a. m., no extrae datos nuevos de la API de GitHub y vuelve a dormir. Para eso no fue construido.
Y cuando lo pensé desde una perspectiva del mundo real, entendí de inmediato. En una empresa real, nadie está ahí sentado en alerta a medianoche esperando ejecutar un script. El oleoducto tiene que funcionar solo. Según un horario. Seguramente. Si alguien está mirando o no.
Aquí es donde entran herramientas como Apache Airflow, Prefect y trabajos cron basados en la nube. Estos no son scripts de Python. Son sistemas que viven en servidores, administran programaciones, manejan fallas, envían alertas cuando algo sale mal y mantienen un historial de cada ejecución.
Honestamente, la programación es el concepto que más me entusiasma profundizar a continuación, porque es donde la ingeniería de datos comienza a sentirse como un trabajo de infraestructura real.
Recorriendo los cambios clave
Permítanme repasar los tres cambios reales que hice en la tubería y lo que hace cada uno.
1. Intercambio de CSV por SQLite
# Antes de df_clean.to_csv(‘github_trending_repos.csv’, index=False) # Después de conn = sqlite3.connect(‘github_repos.db’) df_clean.to_sql(‘repos’, conn, if_exists=’append’, index=False) conn.close()
Guardar en un CSV está bien para un análisis único. Pero un CSV es sólo un archivo de texto. No puede consultarlo fácilmente y no se escala bien a medida que crecen sus datos. SQLite es una base de datos real, lo que significa que puede ejecutar consultas SQL en ella, verificar lo que ya hay dentro y construir sobre ella correctamente. Misma simplicidad, mucha más capacidad.
2. Solucionar el problema de duplicados
cursor.execute(”’ ELIMINAR DE repositorios DONDE URL EN (SELECCIONAR URL DE repos_temp) ”’) cursor.execute(”’ INSERTAR EN repositorios SELECCIONAR * DE repos_temp ”’)
Esta es la solución a la idempotencia. Antes de insertar algo, la canalización verifica si ese repositorio ya existe en la base de datos utilizando su URL como identificador único. Si es así, primero lo elimina y luego inserta la versión nueva. Entonces, no importa cuántas veces se ejecute la canalización, siempre terminará con datos limpios y no duplicados.
3. Datos persistentes en Google Drive
# Antes de conn = sqlite3.connect(‘github_repos.db’) # Después de conn = sqlite3.connect(‘/content/drive/MyDrive/github_repos.db’)
Esta es una línea pero lo cambia todo. En lugar de guardar la base de datos en el entorno temporal de Colab donde desaparece cuando se cierra la sesión, se guarda directamente en Google Drive. Cierra el cuaderno, reinicia el tiempo de ejecución y vuelve mañana. Tus datos todavía están ahí esperándote.
Aquí está el proceso completo actualizado que reúne los tres cambios:
solicitudes de importación importar pandas como pd importar sqlite3 desde fecha y hora importar fecha y hora, timedelta # Extraer ayer = (datetime.now() – timedelta(days=1)).strftime(‘%Y-%m-%d’) url = “https://api.github.com/search/repositories” params = { “q”: f”idioma:python creado:>{yesterday}”, “sort”: “stars”, “order”: “desc”, “per_page”: 30 } respuesta = request.get(url, params=params) data = respuesta.json() # Transformar repositorios = []
para repositorio en datos[‘items’]: repos.append({ “nombre”: repositorio[‘name’]”propietario”: repositorio[‘owner’][‘login’]”estrellas”: repositorio[‘stargazers_count’]”bifurcaciones”: repositorio[‘forks_count’]”idioma”: repositorio[‘language’]”descripción”: repositorio[‘description’]”url”: repositorio[‘html_url’]”created_at”: repositorio[‘created_at’]
}) df = pd.DataFrame(repositorios) df_clean = df.dropna(subconjunto=[‘description’]) df_clean = df_clean.copia() df_clean[‘viral’] =df_clean[‘stars’].apply(lambda x: ‘Sí’ si x > 50000 else ‘No’) df_clean = df_clean.sort_values(‘stars’, ascending=False).reset_index(drop=True) # Cargar conn = sqlite3.connect(‘/content/drive/MyDrive/github_repos.db’) cursor = conn.cursor() cursor.execute(” CREAR TABLA SI NO EXISTE repositorios (nombre TEXTO, propietario TEXTO, estrellas INTEGER, bifurcaciones INTEGER, idioma TEXTO, descripción TEXTO, url TEXTO, creado_en TEXTO, viral TEXTO, cargado_en TEXTO) ”’) df_clean[‘loaded_at’] = datetime.now().strftime(‘%Y-%m-%d’) df_clean.to_sql(‘repos_temp’, conn, if_exists=’replace’, index=False) cursor.execute(”’ ELIMINAR DE repositorios DONDE URL EN (SELECCIONAR URL DE repos_temp) ”’) cursor.execute(”’ INSERTAR EN repositorios SELECCIONAR * DE repos_temp ”’) conn.commit() conn.close() print(“Pipeline completo. Duplicados manejados.”)
Entonces, ¿qué es realmente la ingeniería de datos?
Si me hubieran preguntado qué era la ingeniería de datos después de construir mi primer canal ETL, habría dicho que se trataba principalmente de escribir guiones. Extraer, transformar, cargar. Repetir. Así se veía desde fuera.
Pero después de impulsar ese oleoducto más allá y verlo romperse de tres maneras diferentes, ahora lo pienso de manera diferente. La ingeniería de datos consiste en construir sistemas que sean confiables, no solo scripts que se ejecuten. Hay una diferencia. Un guión hace lo que le dices, una vez, cuando lo dices. Un sistema maneja fallas, recuerda lo que ya ha hecho, conserva los datos más allá de una sola sesión y se ejecuta según un cronograma sin que nadie lo observe.
Idempotencia, persistencia, programación. Ninguno de estos conceptos apareció cuando estaba ejecutando mi canalización en un cuaderno. Sólo se revelaron cuando intenté hacerlo funcionar como algo real.
Y en una empresa real, no puede permitirse el lujo de equivocarse. Los datos que produce su canalización se utilizan para tomar decisiones. Si está lleno de duplicados, desaparece de la noche a la mañana o solo se ejecuta cuando alguien recuerda presionar un botón, eso no es una canalización de datos. Eso es una responsabilidad.
Todavía tengo mucho que aprender. La programación es el siguiente muro hacia el que camino. Pero ahora estoy caminando hacia ello sabiendo que la ingeniería de datos nunca se trató solo de escribir guiones. Sólo tuve que romper algunas cosas para ver eso.
Esta es una serie de ingeniería de datos en curso. Sígueme mientras documento cada paso del viaje, incluidas las partes que no van bien.
Conéctese conmigo en LinkedIn, YouTube y Twitter.