Paralelizando Python en Spark Opciones de concurrencia con Pandas

Parallelizando Python en Spark Opciones de concurrencia con Pandas

Aproveche los beneficios de Spark al trabajar con Pandas

En mi rol anterior, pasé algún tiempo trabajando en un proyecto interno para predecir el uso futuro del espacio de almacenamiento en disco para nuestros clientes de Servicios Gestionados en miles de discos. Cada disco está sujeto a sus propios patrones de uso y esto significa que necesitamos un modelo de aprendizaje automático separado para cada disco que tome datos históricos para predecir el uso futuro en cada disco. Si bien realizar esta predicción y elegir el algoritmo correcto para el trabajo es un desafío en sí mismo, realizarlo a gran escala tiene sus propios problemas.

Para aprovechar una infraestructura más sofisticada, podemos buscar alejarnos de las predicciones secuenciales y acelerar la operación del pronóstico paralelizando la carga de trabajo. Esta publicación de blog tiene como objetivo comparar las UDF de Pandas y el módulo ‘concurrent.futures’, dos enfoques de procesamiento simultáneo, y determinar los casos de uso para cada uno.

El Desafío

Pandas es un paquete fundamental en Python para trabajar con conjuntos de datos en el espacio de análisis. Al trabajar con DataFrames, podemos perfilar datos y evaluar la calidad de los mismos, realizar análisis exploratorios de datos, construir visualizaciones descriptivas de los datos y predecir tendencias futuras.

Aunque esto ciertamente es una gran herramienta, la naturaleza de un único hilo de Python hace que no se escale bien cuando se trabaja con conjuntos de datos más grandes, o cuando es necesario realizar el mismo análisis en múltiples subconjuntos de datos.

• Sam Altman despedido como CEO de OpenAI
• LastMile AI lanza AiConfig un marco de desarrollo de aplicaciones de IA basado en configuración de código abierto y compatible con el control de fuente.
• Conoce a SPHINX Un modelo de lenguaje grande multi-modal y versátil (MLLM) con una combinación de tareas de entrenamiento, dominios de datos y embebimientos visuales.

En el mundo del big data, esperamos un enfoque más sofisticado, ya que tenemos un enfoque adicional en la escalabilidad para mantener un excelente rendimiento. Spark, entre otros lenguajes, nos permite aprovechar el procesamiento distribuido para ayudarnos a procesar estructuras de datos más grandes y complicadas.

Antes de profundizar en este ejemplo específico, podemos generalizar algunos casos de uso que resumen la necesidad de concurrencia en el procesamiento de datos:

• Aplicar transformaciones uniformes a varios archivos de datos
• Pronosticar valores futuros para varios subconjuntos de datos
• Ajustar hiperparámetros de modelos de aprendizaje automático y seleccionar la configuración más eficiente

Al intensificar nuestro requisito para realizar cargas de trabajo como las sugeridas anteriormente y en nuestro caso, el enfoque más sencillo en Python y Pandas es procesar estos datos de manera secuencial. Para nuestro ejemplo, ejecutaríamos el flujo anterior para un disco a la vez.

Los Datos

En nuestro ejemplo, tenemos datos para miles de discos que muestran el espacio libre registrado a lo largo del tiempo y queremos predecir los valores futuros de espacio libre para cada uno de los discos.

Para tener una idea más clara, he proporcionado un archivo csv que contiene 1,000 discos, cada uno con un mes de datos históricos de espacio libre medido en GB. Esto es lo suficientemente grande como para ver el impacto de los diferentes enfoques para predecir a gran escala.

Para un problema de series de tiempo como este, buscamos utilizar datos históricos para predecir tendencias futuras y queremos entender qué algoritmo de aprendizaje automático (ML) será el más apropiado para cada disco. Herramientas como AutoML son geniales para esto cuando se busca determinar el modelo adecuado para un conjunto de datos, pero aquí estamos tratando con 1,000 conjuntos de datos, por lo que es excesivo para nuestra comparación.

En este caso, limitaremos el número de algoritmos que queremos comparar a dos y veremos cuál es el mejor modelo para usar, para cada disco, utilizando el Error Cuadrático Medio (RMSE) como métrica de validación. Puede encontrar más información sobre RMSE aquí. Estos algoritmos son:

• Regresión lineal
• Fbprophet (ajuste de los datos a una línea más compleja)
• Modelo de pronóstico de series de tiempo de Facebook.
• Diseñado para predicciones más complejas con hiperparámetros para la estacionalidad.

Ya tenemos todos los componentes listos ahora si quisiéramos predecir el espacio libre futuro de un solo disco. Las acciones seguirían el siguiente flujo:

Ahora queremos escalar esto, realizando este flujo para múltiples discos, 1,000 en nuestro ejemplo.

Como parte de nuestra revisión, compararemos el rendimiento de calcular los valores de RMSE para los diferentes algoritmos en diferentes escalas. Para ello, he creado un subconjunto de los primeros 100 discos para simular esto.

Esto debería brindar algunas ideas interesantes sobre el rendimiento en conjuntos de datos de diferentes tamaños, realizando operaciones de complejidad variable.

Introducción a la concurrencia

Python es famoso por ser de un solo hilo y, por lo tanto, no aprovecha todos los recursos de cómputo disponibles en un momento dado.

Como resultado, vi tres opciones:

1. Implementar un bucle for para calcular las predicciones secuencialmente, utilizando el enfoque de un solo hilo.
2. Usar el módulo futures de Python para ejecutar varios procesos a la vez.
3. Utilizar UDFs (funciones definidas por el usuario) de Pandas para aprovechar la computación distribuida en PySpark mientras mantenemos nuestra sintaxis de Pandas y paquetes compatibles.

Quería hacer una comparación bastante exhaustiva en diferentes condiciones del entorno, por lo que he utilizado un clúster de Databricks de un solo nodo y otro clúster de Databricks con 4 nodos de trabajo para aprovechar Spark en nuestro enfoque de UDF de Pandas.

Seguiremos el siguiente enfoque para evaluar la idoneidad de los modelos de Regresión Lineal y fbprophet para cada disco:

• Dividir los datos en conjuntos de entrenamiento y prueba
• Utilizar el conjunto de entrenamiento como entrada y predecir sobre las fechas del conjunto de prueba
• Comparar los valores predichos con los valores reales en el conjunto de prueba para obtener un puntaje de Error Cuadrático Medio (RMSE)

Vamos a devolver dos cosas en nuestras salidas: un DataFrame modificado con las predicciones, lo que nos dará el beneficio adicional de graficar y comparar los valores predichos vs los valores reales, y un DataFrame que contiene los puntajes de RMSE para cada disco y algoritmo.

Las funciones para hacer esto se ven así:

Vamos a comparar los tres enfoques mencionados anteriormente. Tenemos algunos escenarios diferentes, por lo que podemos completar una tabla con los resultados que estamos recolectando:

Con las siguientes combinaciones:

Método

• Secuencial
• futures
• Pandas UDFs

Algoritmo

• Regresión lineal
• Fbprophet
• Combinado (ambos algoritmos para cada disco) – la forma más eficiente de realizar una comparación.

Modo de clúster

• Clúster de un solo nodo
• Clúster estándar con 4 nodos de trabajo

Número de discos

• 100
• 1000

Los resultados se presentan en este formato en el apéndice de este blog, si deseas echar un vistazo más detallado.

Los métodos

Método 1: Secuencial

Método 2: concurrent.futures

Hay dos opciones para usar este módulo: paralelizar operaciones intensivas en memoria (usando ThreadPoolExecutor) u operaciones intensivas en CPU (ProcessPoolExecutor). Una explicación descriptiva de esto se encuentra en el siguiente blog. Como vamos a trabajar en un problema intensivo en CPU, ProcessPoolExecutor es adecuado para lo que queremos lograr.

Método 3: Pandas UDFs

Ahora vamos a cambiar de marcha y usar Spark y aprovechar la computación distribuida para ayudarnos con nuestra eficiencia. Como estamos utilizando Databricks, la mayoría de nuestra configuración de Spark se hace automáticamente, pero hay algunos ajustes en nuestro manejo general de datos.

Primero, importa los datos a un DataFrame de PySpark:

Vamos a utilizar el UDF map agrupado de Pandas (PandasUDFType.GROUPED_MAP), ya que queremos pasar un DataFrame y devolver un DataFrame. Desde Apache Spark 3.0 no necesitamos declarar explícitamente este decorador más!

Necesitamos separar nuestras funciones de fbprophet, regresión y RMSE para las UDF de Pandas debido a la estructuración del DataFrame en PySpark, pero no requerimos una revisión de código masiva para lograr esto.

Luego podemos usar applyInPandas para producir nuestros resultados.

Nota: los ejemplos anteriores solo están demostrando el proceso para el uso de la regresión lineal por razones de legibilidad. Por favor, consulta el notebook completo para la demostración completa de esto.

Interpretando los resultados

Hemos creado gráficos para los diferentes métodos y diferentes configuraciones de entorno, luego agrupamos los datos por algoritmo y número de discos para una fácil comparación.

Por favor, ten en cuenta que los resultados tabulares se encuentran en el apéndice de esta publicación.

He resumido los aspectos más destacados de estos hallazgos a continuación:

• Como era de esperar, predecir 1,000 discos en comparación con 100 discos es (generalmente) un proceso que consume más tiempo.
• El enfoque secuencial generalmente es el más lento, ya que no puede aprovechar los recursos subyacentes de manera eficiente.
• Las UDF de Pandas son bastante ineficientes en tareas más pequeñas y simples. La sobrecarga de transformar los datos es más costosa; la paralelización ayuda a compensar esto.
• Tanto los enfoques secuenciales como los concurrent.futures desconocen la posibilidad de agrupación disponible en Databricks, lo que implica una pérdida de capacidad de cómputo adicional.

Pensamientos finales

El contexto ciertamente juega un gran papel en qué enfoque es más exitoso, pero dado que Databricks y Spark suelen usarse para problemas de Big Data, podemos ver el beneficio de utilizar UDF de Pandas con esos conjuntos de datos más grandes y complejos que hemos visto aquí hoy.

Usar un entorno Spark para conjuntos de datos más pequeños se puede hacer con la misma eficiencia en una configuración de computación más pequeña (¡y menos costosa!) como se demuestra mediante el uso del módulo concurrent.futures, así que ten esto en cuenta al diseñar tu solución.

Si estás familiarizado con Python y Pandas, ninguno de estos enfoques debería requerir un cambio drástico en la forma de aprender en comparación con el enfoque de bucle secuencial visto en tutoriales para principiantes.

No hemos investigado esto en esta publicación, ya que he encontrado discrepancias e incompatibilidades con la versión actual, pero el reciente módulo pyspark.pandas seguramente será más común en el futuro y es un enfoque a tener en cuenta. Esta API (junto con Koalas, desarrollado por los chicos de Databricks, pero ahora retirado) aprovecha la familiaridad de Pandas con los beneficios subyacentes de Spark.

Para demostrar el efecto que estamos tratando de lograr, solo hemos llegado a ver los valores de RMSE producidos para cada disco, en lugar de predecir un conjunto de valores de series temporales futuras. El marco que hemos establecido aquí se puede aplicar de la misma manera para esto, con lógica para determinar si la métrica de evaluación (junto con otra lógica, como las limitaciones físicas de un disco) es apropiada en cada caso y predecir los valores futuros, cuando sea posible, utilizando el algoritmo determinado.

Como siempre, el notebook se puede encontrar en mi GitHub.

Apéndice

Publicado originalmente en https://blog.coeo.com, adaptado para esta republicación.

We will continue to update Zepes; if you have any questions or suggestions, please contact us!