MLops: Construyendo un pipeline de procesamiento sobre aws con airflow, mlflow y spark: Pt 1
Previamente escribí un post que demuestra con un prototipo sencillo como llevar a cabo la operacionalización de un modelo de machine learning, puedes consultar la publicación aquí.
Sin embargo, existe una brecha importante entre este prototipo y la implementación a nivel empresarial, por lo cual considere importante construir este post.
El objetivo de este publicación es demostrar una arquitectura funcional escalable basada en la nube de aws que funcione como base tecnológica de procesamiento de grandes volúmenes de datos y de modelos de machine learning.
El código de este post se puede descargar del siguiente repositorio de github.
Dada la cantidad de información y conceptos, el contenido se dividirá en dos partes de la siguiente manera:
- En la primera sección se levantará una infraestructura de control (airflow-mlflow) mediante la herramienta cloudformation creando y configurando máquinas virtuales, interfaces de red, roles de seguridad, almacenamiento, etc. Una vez creada la infraestructura de control, se activará un pipeline de procesamiento de datos mediante airflow.
Este pipeline involucra lo siguiente: conversión de archivos de csv a parquet, joins, entrenamiento de un modelo de machine learning así como loggeo del mismo y sus métricas de desempeño en una herramienta de gestión (mlflow). - En la segunda parte se mostrará como llevar a cabo la operacionalización del modelo de machine learning entrenado en la plataforma de orquestación de contenedores de linux kubernetes. Esta parte abarcará la creación y monitoreo del clúster así como la gestión de los microservicios que se ejecuten en él.
Parte 1
Como se mencionó previamente, en esta parte construiremos una arquitectura orientada al procesamiento de datos y machine learning por lo que es importante presentar el diagrama de arquitectura y conceptos. De igual manera primero se explicarán los conceptos clave y posteriormente el proceso.
Conceptos:
- AWS: Plataforma de nube de la empresa Amazon la cual contiene una gran cantidad de servicios como máquinas virtuales, data lakes, procesamiento en streaming, internet de las cosas entre otros. Este funciona bajo un esquema de autoservicio y pago sobre demanda. A continuación les presento una tabla comparativa entre los distintos proveedores de nube en cuanto su participación de mercado.
Airflow: Plataforma open source para orquestar procesos de ciencia de datos completos en cuanto a sus secuencia de ejecución, calendarización y manejo de errores. Airflow esta disponible como un módulo de python. Como dato, Airflow es intensamente usado en GCP como orquestador de procesos de analítica.
MLflow: Es un proyecto open source que facilita la gestión de modelos de machine learning incluyendo la experimentación, la reproducibilidad así como el despliegue de los mismos. El proyecto fue desarrollado por el equipo creador de spark y está disponible como un módulo de python.
Mayor información sobre MLflow puede ser encontrado en el siguiente link.
Postgres: Base de datos relacional open source de gran potencia que se utilizara como backend de airflow y mlflow.
Spark: Sistema de computo distribuido basado en el paradigma mapreduce de hadoop, sin embargo una diferencia clave es que recae de mayor manera en el uso de memoria resultando ser mucho mas eficiente para el procesamiento de gran escala.
Elastic Cloud Compute (EC2): Servicio de aws que permite la gestión de máquinas virtuales, presenta una gran variedad de imágenes base, configuraciones de seguridad y redes. EC2 es la base utilizada por la gran mayoría de los servicios ofrecidos por la plataforma.
Almacenamiento de archivos (s3): Es el servicio principal de almacenamiento de archivos en aws, teniendo diversas configuraciones y gestiones del ciclo de vida de los archivos.
Cloudformation: Herramienta de gestión de infraestructura como código de la plataforma en la nube Amazon Web Services que permite mediante archivos json o yaml definir arquitecturas completas en la nube. Para gestionar infraestructura multi-nube existen otras herramientas como Terraform.
Ya que contamos con una base teórica de las componentes que se utilizarán en el proyecto, comencemos.
Comencemos
Paso 1: Configurar entorno inicial
a. Crear una cuenta con AWS. Como hacerlo consultar el siguiente link.
b. Descargar y configurar cliente de terminal de aws con cuenta para levantar infraestructura en la region us-west-2. Como hacerlo consultar el siguiente link.
c. Crear una llave del servicio de máquinas virtuales ec2 con nombre airflow_key_pair, descargarla y tenerla disponible en una carpeta (la requeriremos mas adelante). Para mas información sobre como crear un par de llaves consultar el siguiente link.
d. Una vez configurado el cliente de aws, descargar el código de github:
$ git clone https://github.com/al102964/airflow-mlflow-awse. Manualmente crear una bucket en s3 con el nombre que se desee y cargar el archivo bootstrap.sh que se encuentra en la raíz del repo de github. Este archivo contiene los módulos de python adicionales que necesita el clúster de spark para loggear modelos en mlflow.
En mi caso lo cargue en el bucket s3://al102964-bucket1/bootstrap.sh
f. Manualmente modificar la línea 60 del archivo: dags/airflowlib/emr_lib.py
Cambiar: s3://al102964-bucket1/bootstrap.sh
Por: s3://nombre-de-su-bucket/boostrap.sh
Paso 2: Levantar entorno de cloudformation:
Cloudformation tomara el archivo airflow-mlflow.yaml como los planos para la construcción de la infraestructura de control del proyecto. AWS además de contar con el servicio de manejo de infraestructura como código cuenta con un diseñador llamado Cloudformation designer que permite diseñar infraestructura de una manera semejante a hacer un diagrama en powerpoint para posteriormente instanciarla.
AWS llama “stack” a las arquitecturas creadas mediante cloudformation.
Para iniciar con la construcción de la arquitectura de control, ejecutamos los siguientes comandos en nuestra terminal:
$ cd airflow-mlflow-aws/cloudformation$ aws cloudformation create-stack --template-body file://airflow-mlflow.yaml --stack-name airflow-mlflow --parameters ParameterKey=KeyName,ParameterValue=airflow_key_pair --capabilities CAPABILITY_NAMED_IAM
Lo cuál nos regresará lo siguiente:
Podemos observar que nos regresa un StackId que es el identificador interno de AWS de la arquitectura que estamos creando.
Posteriormente ingresamos a la consola web de cloudformation para verificar la creación de nuestro stack:
Después de esto, esperamos aproximadamente 7 minutos en lo que levanta todos los componentes del stack.
Como resultado del stack creado, cloudformation nos arrojará dos DNSs, uno para el servidor de mlflow y otro para el servidor de airflow.
Sin embargo, a pesar de que en cloudformation nos indique que esta listo el stack, tardará aún aproximadamente 2 minutos los dos servicios en levantar. Le damos clic a ambos URLs y refrescamos hasta que hayan levantado los servicios completamente.
Una vez listos los servicios de airflow y mlflow, únicamente nos faltan dos pasos para terminar de levantar el entorno.
- Cambiar referencia del servidor de mlflow contenida en el archivo /root/airflow/dags/transform/linear_regression.py.
Cambiar la línea 18 del archivo
sustituir: mlflow.set_tracking_uri(“http://cambiame:5000")
por: mlflow.set_tracking_uri(“http://{mlflow-dns-server}:5000”)
Primeramente ingresamos al servidor de airflow mediante el siguiente comando:
$ ssh -i "ubicacion_llave/airflow_key_pair.pem" ec2-user@url-airflowEn mi caso particular, ejecuto el siguiente comando:
$ ssh -i /home/personal/blog/aws/airflow_key_pair.pem ec2-user@ec2-54-244-104-66.us-west-2.compute.amazonaws.com
$ sed 's/cambiame/{mlflow-server-DNS}/g' /root/airflow/dags/transform/linearEn mi caso particular, ejecuto la siguiente sentencia:
$ sed -i 's/cambiame/ec2-54-202-241-163.us-west-2.compute.amazonaws.com/g' /root/airflow/dags/transform/linear_regression.py2. Iniciar el calendarizador de tareas
Una vez loggeados al servidor de airflow, iniciamos el scheduler mediante los siguientes comandos:
$ sudo su
$ cd source ~/.bash_profile
$ cd ~/airflow
$ airflow schedulerLo cuál nos arrojará lo siguiente:
Una vez iniciado el calendarizador de tareas refrescamos la consola web de airflow, observamos que ha desaparecido la advertencia que indica que el calendarizador no esta corriendo y que esta cargado un DAG llamado transform_movielens.
Acerca de este pipeline o DAG
En este pipeline (DAG en nomenclatura de spark y airflow) se toman los datasets del proyecto publico movielens que es una recopilación de data acerca de películas publicas en donde aparecen características como los ratings de las películas, categorías, información de usuarios, etc. https://grouplens.org/datasets/movielens/
A partir de esta información base, se podrían plantear dos escenarios de machine learning:
- Motor de recomendación para películas de interés para los usuarios en base a los ratings que han hecho previamente.
De ser de interés los motores de recomendación checa la siguiente liga. - Predecir el rating de películas mediante una regresión lineal tomando como variables las categorías a las que pertenece. Ej: Que tanto rating tendrá una película si pertenece a las categorías de acción, aventura, violencia, etc.
Escogí el segundo escenario de machine learning (predicción de ratings).
Como esta construido?
Damos clic en el hipervínculo de transform_movielens y como esta compuesto este pipeline.
Podemos observar que esta compuesto por 11 etapas:
- Iniciar construcción de clúster de spark emr
- Esperar a que el clúster este arriba
3–8: Convertir archivos base de csv a parquet
9. Creación de tabla maestra insumo para el algoritmo de machine learning
10. Ejecución del algoritmo de machine learning (regresión lineal) y loggeo de métricas y modelo entrenado en mlflow.
11. Terminar el clúster
La especificación del DAG se encuentra dentro del archivo movielens_dag.py que contiene los siguiente: Todas las instrucciones para cada uno de los pasos del pipeline, la creación del clúster de spark mediante el servicio de emr, 6 etapas de compresión de archivos csv a parquet, un proceso de joins de 3 de estos archivos parquet y un algoritmo de machine learning.
A continuación pongo como ejemplo el código de uno de los 6 pasos del pipeline que convierte archivos csv en parquet. Los otros 5 pasos con prefijo transform hacen lo mismo pero para archivos distintos.
Una vez que tenemos un entendimiento de la lógica del dag así como de su código, procedemos a iniciarlo. Damos clic en Off y cambiara a estatus On y damos clic en Trigger DAG, lo cual iniciara el DAG inmediatamente.
Después de unos minutos, el DAG debe haber concluido satisfactoriamente, lo cual se puede observar con lo contornos verdes obscuros de cada una de las etapas.
Dado que concluyó satisfactoriamente la ejecución de este pipeline, nos dirigimos a MLflow para ver los resultados estadísticos del modelo.
En esta vista podemos observar que está loggeada la ejecución del proceso, sus parámetros de ejecución: elasticNetParam, maxIter, regParam y sus métricas de desempeño r2 = 0.8431 y rmse = 0.19232 (Lo cual es bastante bueno para una regresión).
Para obtener más detalle de la ejecución del proceso, damos clic en el hipervínculo de la ejecución.
Posteriormente en el url que accedimos, bajamos a la parte inferior para observar que el modelo se loggeo exitosamente en MLflow, con la especificación de MLmodel y el ambiente de conda.yaml, lo que nos permitirá en la parte 2 instanciar el modelo entrenado sobre un clúster de kubernetes.
Consideraciones
Esta publicación presenta los elementos esenciales de una arquitectura de procesamiento de big data y machine learning.
La siguiente publicación será la continuación de este post en donde se pondrá en operación el modelo entrenado sobre una infraestructura de kubernetes.
Sin embargo es importante resaltar algunos aspectos relevantes faltantes de esta sección:
- Diseño mas robusto en cuanto a networking de cada una de las componentes de la arquitectura (security groups y subnets).
- Organización de modelos de machine learning en experimentos de mlflow.
- Dimensionamiento de volumetría para utilizar la infraestructura adecuada para el caso de uso en cuestión así como optimización de procesos mediante el análisis de los diagramas de Gantt tanto en Airflow como en el spark History Server.
- Habilitamiento de configuración de auto-scaling de los servidores de airflow y mlflow en caso de necesitar alta disponibilidad.
- Calendarización de pipelines en base a eventos a modo de ejemplo: que lleguen nuevos archivos al data lake (s3).
- Gestión de errores centralizado para la mejora operativa de los procesos mediante herramientas tipo Sentry.
Conclusiones
Dado que el objetivo de este post es demostrar una arquitectura en la nube escalable, lo ideal es que se tome como punto de partida los procesos necesarios en cualquiera de sus componentes adecuados para el diseño de otras arquitecturas de procesamiento.
Recomiendo echar un vistazo a arquitecturas adicionales para diseñar la arquitectura más eficiente posible para cada caso de uso.
A continuación recomiendo arquitecturas complementarias
Referencias adicionales
Se consultó material de un post de una excelente ingeniera de Amazon Binal Jhaveri, tomé algunos elementos de su trabajo y lo complementé con elementos adicionales.
