Programar y registrar ejecuciones del pipeline

Un pipeline que solo se ejecuta cuando un desarrollador abre manualmente un notebook no aporta ningún valor empresarial más allá de la primera ejecución. Para ejecutarse automáticamente cada día, el pipeline debe estructurarse como un script de Python ejecutable desde la línea de comandos: python pipeline.py. Esto requiere un punto de entrada if __name__ == '__main__':, análisis de argumentos de la línea de comandos y un registro adecuado; estos son los tres pilares de un script de producción.

El módulo integrado logging de Python es la herramienta adecuada para los registros del pipeline, no las instrucciones print(). Configure un logger con salida tanto en la consola como en un archivo mediante logging.basicConfig(). Registre el progreso normal en el nivel INFO y los fallos en el nivel ERROR. Los registros basados en archivos persisten después de que finaliza el proceso, lo cual es esencial para depurar ejecuciones programadas que nadie estaba supervisando.

Registre siempre la hora de inicio, la hora de finalización y el tiempo transcurrido de una ejecución del pipeline. Esto establece una referencia: si normalmente el pipeline tarda 45 segundos y hoy tardó 8 minutos, algo ha cambiado; quizá el archivo de entrada sea 10 veces más grande o una consulta a la base de datos se esté ejecutando lentamente. Las entradas de registro de inicio y finalización con marca de tiempo hacen que esta comparación sea trivial utilizando únicamente el archivo de registro.

Registre el número de filas que entran y salen de cada paso de transformación. Un registro claro tiene este aspecto: extract: 50,000 rows → drop_nulls: 49,200 rows → filter: 47,800 rows → output: 47,800 rows. Esta traza muestra de inmediato cuántas filas se eliminaron en cada paso y si las cifras son las esperadas. Las eliminaciones anómalas aparecen como saltos en los recuentos registrados.

cron es el programador estándar de Unix para trabajos recurrentes. Edite el crontab con crontab -e y añada una línea que especifique cuándo ejecutar el script. El formato es: minuto hora día mes día_de_la_semana comando. Un pipeline que deba ejecutarse todos los días a las 6:00 utiliza 0 6 * * * /usr/bin/python /path/to/pipeline.py. Utilice siempre rutas absolutas en las entradas de cron, porque cron se ejecuta en un entorno mínimo sin la configuración de PATH de su shell.

La biblioteca schedule proporciona una forma basada exclusivamente en Python de ejecutar trabajos a intervalos especificados sin utilizar cron. Resulta útil en entornos donde cron no está disponible (Windows) o cuando desea mantener la lógica del programador dentro del propio proceso de Python. Coloque el pipeline en un bucle de trabajo programado y mantenga el proceso activo para ejecutarlo repetidamente.

Un script de pipeline debe devolver un código de salida distinto de cero cuando falla, para que el programador sepa que el trabajo ha fallado. Incluya la ejecución principal en un bloque try/except y llame a sys.exit(1) en caso de error. cron, Jenkins y Airflow comprueban el código de salida: un código distinto de cero activa una alerta, una nueva ejecución o una notificación. Una excepción no controlada que no establezca el código de salida puede pasar desapercibida para la supervisión automatizada.

Después de una ejecución correcta, escriba un pequeño archivo de resumen JSON junto a la salida. Incluya la marca de tiempo de la ejecución, el número de filas de entrada, el número de filas de salida, las filas eliminadas y el tiempo transcurrido. Los sistemas de supervisión y los paneles pueden leer este archivo para seguir la evolución del estado del pipeline a lo largo del tiempo. Un panel que muestre las filas de salida de los últimos 30 días facilita detectar el día en que una fuente de datos empezó a entregar menos registros.

Si un pipeline programado se activa dos veces por accidente, no debe corromper la salida. Diseñe el paso de carga para que sea idempotente: utilice un nombre de archivo de salida con la fecha o sobrescriba la misma salida con el resultado más reciente. Para las cargas en bases de datos, utilice if_exists='replace' o un patrón UPSERT. Nunca utilice el modo append sin un paso de eliminación de duplicados, ya que cada ejecución programada añadirá filas duplicadas a la tabla de salida.

En los pipelines de los que dependen las operaciones empresariales, que no haya ninguna señal después de un fallo es peligroso. Configure una alerta sencilla: si el archivo de resumen de ejecución no se actualiza dentro del intervalo esperado, envíe un correo electrónico o un mensaje de Slack. smtplib de Python puede enviar un correo electrónico cuando se produce un fallo, o puede usar un webhook para publicar un mensaje en Slack. Genere una alerta inmediatamente ante un código de salida 1 o 2 para que la persona analista sepa que la actualización diaria no se ha realizado antes de que la empresa lo detecte.

Combine todas las piezas —parseo de argumentos, configuración del registro, resumen de ejecución, gestión de errores y códigos de salida— en un script de pipeline completo. Puede incorporar este script en cualquier entorno, indicarle un archivo de configuración y programarlo con cron o con cualquier orquestador de flujos de trabajo. En cada ejecución genera un archivo de registro fechado, un resumen de ejecución y un archivo de salida fechado, lo que hace que cada ejecución sea completamente auditable y reproducible de forma independiente.

En esta lección ha aprendido: estructurar un pipeline como un script de línea de comandos con parseo de argumentos y registro, programar tareas con cron y gestionar fallos mediante códigos de salida distintos de cero y alertas, y escribir archivos de resumen de ejecución y diseñar pasos de carga idempotentes para una ejecución automatizada fiable. ¡Enhorabuena por completar el itinerario de Análisis de datos: Pandas y NumPy!