Tutorial de Metaflow para principiantes: Construye y Escala Flujos de Trabajo de Datos

Configurar Metaflow

Metaflowsugiere que los usuarios instalen Python 3 en lugar de Python 2.7 para los nuevos proyectos. La documentación afirma que "Python 3 tiene menos errores y está mejor soportado que el obsoleto Python 2.7".

El siguiente paso es crear un entorno virtual para gestionar las dependencias de tu proyecto. Ejecuta el siguiente comando para hacerlo: 

python -m venv venv
source venv/bin/activate

Esto creará y activará un entorno virtual. Una vez activado, estás listo para instalar Metaflow. 

Metaflow está disponible como paquete Python para MacOS y Linux. La última versión se puede instalar desde el repositorio Metaflow de Github o desde PyPi ejecutando el siguiente comando: 

pip install metaflow

Por desgracia, en el momento de escribir esto, Metaflow no ofrece soporte nativo para usuarios de Windows. Sin embargo, los usuarios de Windows 10 pueden utilizar WSL (Windows Subsystem for Linux) para instalar Metaflow, que les permite ejecutar un entorno Linux dentro de su sistema operativo Windows. Consulta la documentación para obtener una guía paso a paso sobre la instalación de Metaflow en Windows 10. 

Integración con AWS (Opcional)

Metaflow se ofrece sin fisuras con AWS, lo que permite a los usuarios de escalar sus flujos de trabajo utilizando la infraestructura de la nube. Para integrar AWS, tendrás que configurar tus credenciales de AWS.

Nota: Estos pasos suponen que ya tienes una cuenta de AWS y la CLI de AWS instalada. Para más detalles, sigue las instrucciones de la documentación de AWS. 

1. Primero, instala la CLI de AWS ejecutando:

pip install awscli

2. Configura AWS ejecutando

aws configure

A partir de aquí, se te pedirá que introduzcas tu ID de clave de acceso de AWS y tu clave de acceso secreta, que son simplemente las credenciales que utiliza la CLI de AWS para autenticar tus solicitudes a AWS. Ten en cuenta que también se te pedirá que introduzcas tu región y el formato de salida. 

Una vez que hayas introducido estos datos, ¡voilá! Metaflow utilizará automáticamente tus credenciales de AWS para ejecutar los flujos de trabajo. 

Crear tu primer flujo de trabajo con Metaflow

Ahora que Metaflow está configurado, es hora de construir tu primer flujo de trabajo. En esta sección, te guiaré a través de los conceptos básicos para crear un flujo, ejecutarlo y comprender cómo se organizan las tareas y los pasos en Metaflow. 

Al final de esta sección, tendrás un flujo de trabajo que procesa datos y realiza operaciones sencillas. ¡Vamos!

Visión general de un flujo Metaflow

Metaflow utiliza el paradigma del flujo de datos, que representa un programa como un grafo dirigido de operaciones. Este enfoque es ideal para construir pipelines de procesamiento de datos, especialmente en el aprendizaje automático.

En Metaflow, el gráfico de operaciones se denomina flujo. Un flujo consiste en una serie de tareas divididas en pasos. Observa que cada paso puede considerarse una operación representada como un nodo, con transiciones entre pasos que actúan como las aristas del grafo.

Una transición lineal básica Metaflow | Fuente: Documentación Metaflow

En Metaflow existen algunas reglas estructurales para los flujos. Por ejemplo, todo flujo debe incluir un pasoinicial y un pasofinal. Cuando se ejecuta un flujo, lo que se conoce comoejecución , comienza con el paso inicial y se considera que tiene éxito si llega al paso final sin errores .

Lo que ocurra entre los pasos inicial y final depende totalmente de ti, como verás en el siguiente segmento. 

Escribir tu primer flujo

Aquí tienes un flujo sencillo para empezar. Nota: El código puede ejecutarse en DataLab.

from metaflow import FlowSpec, step

class MyFirstFlow(FlowSpec):
    @step
    def start(self):
        print("Starting the flow!")
        self.data = [1, 2, 3, 4, 5]  # Example dataset
        self.next(self.process_data)
    @step
    def process_data(self):
        self.processed_data = [x * 2 for x in self.data]  # Simple data processing
        print("Processed data:", self.processed_data)
        self.next(self.end)
    @step
    def end(self):
        print("Flow is complete!")

if __name__ == '__main__': 
    MyFirstFlow()

En este flujo:

• El paso start() inicializa el flujo de trabajo y define un conjunto de datos.
• El paso process_data() procesa los datos duplicando cada elemento.
• El paso end() completa el flujo.

Cada paso utiliza el decorador @step, y tú defines la secuencia de flujo utilizando self.next() para conectar los pasos.

Ejecutar tu flujo

Después de escribir tu flujo, guárdalo como my_first_flow.py. Ejecútalo desde la línea de comandos utilizando

py -m my_first_flow.py run

En Metaflow 2.12 se ha añadido una nueva función que permite a los usuarios desarrollar y ejecutar flujos en cuadernos. 

Para ejecutar un flujo en una celda definida, sólo tienes que añadir la línea NBRunner en la última línea de la misma celda. Por ejemplo:

from metaflow import FlowSpec, step, NBRunner

class MyFirstFlow(FlowSpec):
    @step
    def start(self):
        print("Starting the flow!")
        self.data = [1, 2, 3, 4, 5]  # Example dataset
        self.next(self.process_data)
    @step
    def process_data(self):
        self.processed_data = [x * 2 for x in self.data]  # Simple data processing
        print("Processed data:", self.processed_data)
        self.next(self.end)
    @step
    def end(self):
        print("Flow is complete!")

run = NBRunner(MyFirstFlow).nbrun()

Si aparece el siguiente error

"Metaflow no pudo determinar tu nombre de usuario basándose en las variables de entorno ($USERNAME etc.)"

Añade lo siguiente a tu código antes de la ejecución de Metaflow:

import os
if os.environ.get("USERNAME") is None:
    os.environ["USERNAME"] = "googlecolab"

En ambos casos, Metaflow ejecutará el flujo paso a paso. Es decir, mostrará la salida de cada paso en el terminal de la siguiente manera: 

La salida del código anterior | Fuente: Imagen del autor

Conceptos básicos de Metaflow

Comprender los conceptos básicos de Metaflow es esencial para construir flujos de trabajo de datos eficientes y escalables. En esta sección, trataré tres conceptos fundamentales: 

• Pasos y ramificaciones
• Artefactos de datos
• Versionado 

Estos elementos forman la columna vertebral de la estructura y ejecución de los flujos de trabajo de Metaflow, permitiéndote gestionar procesos complejos con facilidad.

Pasos y ramificaciones

Hemos tratado brevemente los pasos anteriormente en el artículo, pero en aras de la claridad, volveremos sobre ellos. Lo más importante que hay que entender sobre los flujos de trabajo Metaflow es que se construyen en torno a pasos. 

Los pasos representan cada tarea individual dentro de un flujo de trabajo. En otras palabras, cada paso realizará una operación específica (por ejemplo, carga de datos, procesamiento, modelado, etc.). 

El ejemplo que creamos anteriormente en "Escribir tu primer flujo" era una transformación lineal. Además de los pasos secuenciales, Metaflow también permite a los usuarios ramificar los flujos de trabajo. Los flujos de trabajo de bifurcación te permiten ejecutar varias tareas en paralelo creando rutas de ejecución separadas. 

Un ejemplo de ramificación | Fuente: Documentación Metaflow

La mayor ventaja de una sucursal es el rendimiento. La ramificación significa que Metaflow puede ejecutar varios pasos en varios núcleos de CPU o instancias en la nube. 

Así es como se vería una rama en código: 

from metaflow import FlowSpec, step, NBRunner

class BranchFlow(FlowSpec):
    @step
    def start(self):
        print("Starting the flow!")
        self.data = [1, 2, 3, 4, 5]  # Example dataset
        self.next(self.split)
    @step
    def split(self):
        self.next(self.branch1, self.branch2)
  
    @step
    def branch1(self):
        # Code for branch 1
        print("This is branch 1")
        self.next(self.join)
    
    @step
    def branch2(self):
        # Code for branch 2
        print("This is branch 2")
        self.next(self.join)
  
    @step
    def join(self, inputs):
        # Merging branches back
        print("Branches joined.")
        self.next(self.end)
  
    @step
    def end(self):
        print("Flow is complete!")

run = NBRunner(BranchFlow).nbrun()

Artefactos de datos

Artefactos dedatos son variables que te permiten almacenar y pasar datos entre los pasos de un flujo de trabajo. Estos artefactos persisten la salida de un paso al siguiente: así es como los datos se ponen a disposición de los pasos posteriores. 

Básicamente, cuando asignas datos a self dentro de un paso de una clase Metaflow, los guardas como artefacto, al que luego puede acceder cualquier otro paso del flujo (consulta los comentarios del código).

class ArtifactFlow(FlowSpec):
    @step
    def start(self):
        # Step 1: Initializing data
        print("Starting the flow!")
        self.data = [1, 2, 3, 4, 5]  # Example dataset saved as an artifact
        self.next(self.process_data)
    @step
    def process_data(self):
        # Step 2: Processing the data from the 'start' step
        self.processed_data = [x * 2 for x in self.data]  # Processing artifact data
        print("Processed data:", self.processed_data)
        self.next(self.save_results)
    @step
    def save_results(self):
        # Step 3: Saving the processed data artifact
        self.results = sum(self.processed_data)  # Saving final result as an artifact
        print("Sum of processed data:", self.results)
        self.next(self.end)
    @step
    def end(self):
        # Final step
        print("Flow is complete!")
        print(f"Final result: {self.results}")  # Accessing artifact in final step

¿Por qué los artefactos son un concepto central de Metaflow? Porque tienen varios usos:

• Automatizar la gestión del flujo de datos, eliminando la necesidad de cargarlos y almacenarlos manualmente.
• Habilitar la persistencia (más sobre esto a continuación), lo que significa que permiten a los usuarios realizar análisis más tarde con la API de cliente, visualizar con Cards y reutilizar entre flujos.
• Coherencia en entornos locales y en la nube. Esto elimina la necesidad de transferencias explícitas de datos.
• Permitir a los usuarios inspeccionar los datos antes de los fallos y reanudar las ejecuciones después de arreglar los fallos.

Versionado y persistencia

Metaflow gestiona automáticamente el versionado para tus flujos de trabajo. Esto significa que cada vez que se ejecuta un flujo, se rastrea como una ejecución única. En otras palabras, cada ejecución tiene su propia versión, lo que te permite revisar y reproducir fácilmente las ejecuciones anteriores. 

Metaflow lo hace asignando identificadores únicos a cada ejecución y conservando los datos y artefactos de esa ejecución. Esta persistencia garantiza que no se pierdan datos entre ejecuciones. Los flujos de trabajo anteriores pueden revisarse e inspeccionarse fácilmente, y pueden volver a ejecutarse pasos concretos si es necesario. Como resultado, la depuración y el desarrollo iterativo son mucho más eficientes, y se simplifica el mantenimiento de la reproducibilidad. 

Ejemplo práctico: Entrenar un modelo de aprendizaje automático

En esta sección, te guiaré en el uso de Metaflow para entrenar un modelo de aprendizaje automático. Aprenderás a:

• Definir un flujo de trabajo que cargue datos
• Entrenar un modelo de aprendizaje automático 
• Sigue los resultados

Al final, comprenderás mejor cómo utilizar Metaflow para estructurar y ejecutar eficientemente flujos de trabajo de aprendizaje automático. ¡Vamos!

Para empezar, crearemos un flujo básico que cargue un conjunto de datos, realice el entrenamiento y emita los resultados del modelo. 

Nota: El código se puede ejecutar en DataLab.

from metaflow import FlowSpec, step, Parameter, NBRunner

class TrainModelFlow(FlowSpec):
    @step
    def start(self):
        # Load and split the dataset
        print("Loading data...")
        self.data = [1, 2, 3, 4, 5]  # Replace with actual data loading logic
        self.labels = [0, 1, 0, 1, 0]  # Replace with labels
        self.next(self.train_model)
	   
    @step
    def train_model(self):
        # Training a simple model (e.g., linear regression)
        print("Training the model...")
        self.model = sum(self.data) / len(self.data)  # Replace with actual model training
        print(f"Model output: {self.model}")
        self.next(self.end)
	   
    @step
    def end(self):
        # Final step
        print("Training complete. Model ready for deployment!")f

En este código, definimos tres pasos:

1. start(): Carga y divide el conjunto de datos. En un escenario real, cargarías datos desde una fuente real (por ejemplo, un archivo o una base de datos).
2. train_model(): Simula el entrenamiento de un modelo. Aquí se realiza un simple cálculo de la media en lugar de un algoritmo real de aprendizaje automático, pero puedes sustituirlo por cualquier código de entrenamiento que necesites.
3. end(): Marca el final del flujo y significa que el modelo está listo para su despliegue.

Una vez que hayas definido el flujo, puedes ejecutarlo utilizando el siguiente comando:

run = NBRunner(TrainModelFlow)
run.nbrun()

Ten en cuenta que este código sólo funciona en cuadernos (todo el código debe estar en una celda).

Si quieres ejecutar este código como un script, elimina los comandos NBRunner y adjunta lo siguiente al final de tu script, y guárdalo (por ejemplo, "metaflow_ml_model.py"): 

if __name__ == "__main__": 
    TrainModelFlow()

A continuación, para ejecutar el script, navega hasta la línea de comandos y ejecuta el siguiente comando:

py -m metaflow_ml_model.py 

Metaflow rastrea automáticamente cada ejecución y te permite visualizar los resultados a través de la interfaz de usuario de Metaflow. 

Buenas prácticas para utilizar Metaflow

Entonces, ¿cómo podemos aprovechar al máximo las funciones de Metaflow? He aquí algunas buenas prácticas que pueden ayudarte a conseguir esta hazaña, optimizando al mismo tiempo tus flujos de trabajo: 

Empieza con flujos pequeños

Si eres nuevo en Metaflow, empieza con flujos de trabajo sencillos para familiarizarte con su API. Empezar poco a poco te ayudará a entender cómo funciona el marco y a ganar confianza en sus capacidades antes de pasar a proyectos más complejos. Este enfoque reduce la curva de aprendizaje y garantiza que tus cimientos sean sólidos.

Utiliza la interfaz de usuario de Metaflow para depurar

Metaflow incluye una potente interfaz de usuario que puede ser extremadamente útil para depurar y seguir tus flujos de trabajo. Utiliza la interfaz de usuario para supervisar las ejecuciones, comprobar los resultados de los pasos individuales e identificar cualquier problema que pueda surgir. Visualizar tus datos y registros facilita la identificación y solución de problemas durante la ejecución de tu flujo.

Aprovechar AWS para la escalabilidad

Cuando instalas Metaflow por primera vez, funciona en modo local. En este modo, los artefactos y metadatos se guardan en un directorio local, y los cálculos se ejecutan utilizando procesos locales. Esta configuración funciona bien para uso personal, pero si tu proyecto implica colaboración o grandes conjuntos de datos, es aconsejable configurar Metaflow para que utilice AWS para una mejor escalabilidad. Lo bueno aquí es que Metaflow proporciona una excelente integración con AWS.

Conclusión

En este tutorial, exploramos cómo empezar con Metaflow, desde la instalación hasta la construcción de tu primer flujo de trabajo de ciencia de datos. Cubrimos los conceptos básicos, como la definición de pasos, el uso de artefactos de datos para pasar datos entre pasos, y el versionado para rastrear y reproducir ejecuciones. También recorrimos un ejemplo práctico de entrenamiento de un modelo de aprendizaje automático, que demostró cómo definir, ejecutar y supervisar tu flujo de trabajo. Por último, abordamos algunas prácticas recomendadas para ayudarte a sacar el máximo partido de Metaflow. 

Para seguir aprendiendo sobre MLOps, consulta los siguientes recursos: 

• Conceptos MLOps
• MLOps Fundamentals
• MLOps totalmente automatizados