Un Framework para Big Data Optimization Basado en jMetal y Spark

Un Framework para Big Data Optimization Basado en jMetal y Spark Crist´obal Barba-Gonz´alez1, Antonio J. Nebro1, Jos´e Garc´ıa-Nieto1, Jos´e A. Cordero2, Juan J. Durillo3, Ismael Navas-Delgado1, and Jos´e F. Aldana-Montes1 1Grupo de investigaci´on Khaos Edificio de investigaci´on Ada Byron Departamento de Lenguajes y Ciencias de la Computaci´on Universidad de M´alaga, Espa˜na. 2Organizaci´on Europea para la Investigaci´on Nuclear (CERN), Suiza 3Distributed and Parallel Systems Group Universidad de Innsbruck, Austria Resumen Las metaheur´ısticas multi-objetivo se han convertido en t´ecnicas muy utilizadas para la resoluci´on problemas complejos de optimizaci´on compuestos de varias funciones objetivo en conflicto entre s´ı. Nos encontramos en la actualidad inmersos en la era del Big Data, por lo que los problemas multi-objetivo que surjan en este contexto cumplir´an algunas de las cinco V’s que caracterizan a las aplicaciones Big Data (volumen, velocidad, variedad, veracidad, valor). Como consecuencia, las metaheur´ısticas deber´an ser capaces de resolver problemas din´amicos, que pueden cambiar en el tiempo debido al procesamiento y an´alisis de diferentes fuentes de datos, que t´ıpicamente ser´an en streaming. En este trabajo presentamos el software jMetalSP, que combina el framework jMetal con Apache Spark. De esta forma, las metaheur´ısticas disponibles en jMetal se pueden adaptar f´acilmente para resolver problemas din´amicos que se alimenten de distintas fuentes de datos en streaming, y que son gestionadas por Spark. Se describe la arquitectura de jMetalSP y se valida mediante un caso de uso realista basado en TSP bi-objetivo con datos abiertos reales de tr´afico de la ciudad de Nueva York. Key words: Big Data, Optimizaci´on Multi-objetivo, Problemas Din´amicos, Metaheur´ısticas, Framework Software, jMetal, Apache Spark 1. Introducci´on Estamos inmersos en la era del Big Data, donde las aplicaciones deben gestionar y analizar ingentes cantidades de datos que no pueden ser procesados mediante el uso de tecnolog´ıas tradicionales de bases de datos. El volumen de datos no es la ´unica caracter´ıstica de las aplicaciones Big Data, ya que tienen que procesar adem´as fuentes de datos heterog´eneas (la mayor´ıa de ellas en streaming), que generalmente generan datos a gran velocidad. Estos datos tambi´en deben ser validados y analizados para producir un valor significativo para el 2 Crist´obal Barba-Gonz´alez et al. usuario final [1]. Estas caracter´ısticas constituyen las denominadas V’s del Big Data: volumen, velocidad, variedad, variabilidad, veracidad y valor [2]. En este contexto, los algoritmos evolutivos y metaheur´ısticas en general van a tener un papel importante en la resoluci´on de lo que denominan problemas de Big Data Optimization. Si nos centramos en los problemas multi-objetivo, es decir, los que tienen dos o m´as funciones contradictorias entre s´ı, ´estos se encuentran en muchas disciplinas, tales como el transporte, la econom´ıa, la medicina, la bioinform´atica, etc., por lo que se prev´e que aparezcan variantes Big Data de estos problemas en un futuro inmediato. En este sentido, va a constituir un reto el adaptar metaheur´ısticas actuales para hacer frente a estos nuevos tipos de problemas [3]. Un aspecto importante ser´a disponer de herramientas potentes que permitan aprovechar la gran cantidad de investigaci´on en optimizaci´on multi-objetivo realizada en los ´ultimos 15 a˜nos, con el fin de poder utilizar algoritmos cl´asicos y modernos en combinaci´on con algunas plataformas de gesti´on de grandes vol´umenes de datos. Otra cuesti´on importante es que muchos problemas multi-objetivo del mundo real tienen objetivos, restricciones y par´ametros que pueden cambiar a lo largo del tiempo. Estos problemas se conocen como problemas de optimizaci´on multiobjetivo din´amicos [4]. De hecho, estos se relacionan de manera natural con el Big Data ya que los cambios en los problemas se pueden deber a datos recibidos en streaming a partir de diferentes fuentes de datos. Nuestra motivaci´on en este art´ıculo es la de presentar jMetalSP, una herramienta software para problemas multi-objetivo din´amicos de optimizaci´on en Big Data, que combina el framework jMetal para optimizaci´on multi-objetivo con metaheur´ısticas [5], con el sistema de cluster computing para Big Data Apache Spark [6]. A la hora de dise˜nar el framework jMetalSP se han tenido en cuenta los siguientes requisitos: La herramienta debe permitir definir y resolver problemas din´amicos de Big Data Optimization sobre clusters Hadoop/Spark. Los algoritmos disponibles en jMetal deben ser f´acilmente adaptables para resolver problemas multi-objetivo din´amicos. Las aplicaciones desarrolladas con jMetalSP deben poder incorporar diferentes fuentes de datos, usando para ello los servicios que ofrece Spark. El framework debe ser f´acil de usar, intentando ocultar en lo posible detalles de bajo de nivel y simplificando la incorporaci´on de las fuentes de datos. jMetalSP ser´a un proyecto Open Source, de forma que est´e libremente disponible para cualquier investigador interesado4. Los comentarios de los usuarios permitir´an mejorarlo y hacer que evolucione. Para validar el funcionamiento de jMetalSP hemos definido una versi´on din´amica y bi-objetivo del problema del viajante de comercio (TSP) [7], que incluye una mezcla de datos reales de tr´afico de la ciudad de Nueva York, con datos simulados tomados mediante la API de Twitter 5y Apache Kafka 6. 4Estar´a disponible en GitHub en esta URL: XX 5Available from URL https://dev.twitter.com/overview/api 6Available at URL http://kafka.apache.org/ Un Framework Basado en jMetal y Spark 3 Figura 1. Architectura de jMetal 5 (diagrama UML de classes). El resto del art´ıculo est´a organizado de la siguiente manera. En la secci´on 2 se describen jMetal y Spark, que son los dos componentes de jMetalSP. La arquitectura del framework se presenta en la secci´on 3. La secci´on 4 describe el caso de estudio. Finalmente, la Secci´on 5 expone las principales conclusiones y las l´ıneas de trabajo futuro. 2. Componentes software jMetalSP es una herramienta software que incluye dos componentes. Por un lado, el framework de optimizaci´on multi-objetivo jMetal, que proporcionar´a la infraestructura de optimizaci´on para implementar tanto los problemas de Big Data Optmization, como los algoritmos din´amicos para resolverlos; por otro lado, el sistema de computaci´on distribuida Spark, que permitir´a gestionar las fuentes de datos en streaming y sacar partido a la potencia de c´alculo del cluster Hadoop utilizado. Adem´as permitir´a almacenar y recuperar datos desde HDFS. 2.1. jMetal jMetal es un framework Open Source basado en Java para la optimizaci´on multi-objetivo con metaheur´ısticas [5]. La arquitectura de jMetal 5, la versi´on en la que se basa jMetalSP, se muestra en la Figura 1. Podemos observar que hay cuatro interfaces principales, que modelan la idea de que un algoritmo (una metaheur´ıstica) resuelve un problema mediante una serie de soluciones que se manipulan con un conjunto de operadores, y los problemas son las entidades responsables de crear y evaluar soluciones. La arquitectura b´asica es lo suficientemente gen´erica como para permitir la flexibilidad necesaria para implementar cualquier metaheur´ıstica. Sin embargo, la mayor´ıa de los algoritmos pertenecen a las subfamilias bien establecidas, tales como los algoritmos evolutivos (EAs), la optimizaci´on mediante enjambre de 4 Crist´obal Barba-Gonz´alez et al. part´ıculas (PSO), la b´usqueda dispersa (SS) y muchas otras. Estas subfamilias se caracterizan por un comportamiento com´un que es compartido por todos los algoritmos que pertenecen a ellas. jMetal 5 ofrece una serie de plantillas que incluyen el comportamiento las subfamilias concretas de algoritmos, por lo que el desarrollo de un algoritmo particular s´olo requiere implementar algunos m´etodos concretos. Un ejemplo de plantilla es la clase AbstractEvolutionryAlgorithm, que incluye la siguiente implementaci´on del m´etodo run(), que reproduce el c´odigo de un algoritmo evolutivo t´ıpico: 1. @Override public void run() { 2. List<S> offspringPopulation; 3. List<S> matingPopulation; 4. population = createInitialPopulation(); 5. population = evaluatePopulation(population); 6. initProgress(); 7. while (!isStoppingConditionReached()) { 8. matingPopulation = selection(population); 9. offspringPopulation = reproduction(matingPopulation); 10. offspringPopulation = evaluatePopulation(offspringPopulation); 11. population = replacement(population, offspringPopulation); 12. updateProgress();} 13. } Algunos algoritmos evolutivos multi-objetivo muy conocidos, como NSGAII [8], SPEA2 [9] o SMS-EMOA [10] se basan en esta plantilla. De hecho, el uso de plantillas facilita la reusabilidad de c´odigo (por ejemplo, para implementar una variante un algoritmo determinado s´olo hay que redefinir los m´etodos que difieren del algoritmo original) y es una caracter´ıstica que es explotada en jMetalSP. 2.2. Apache Spark Apache Spark es un sistema de computaci´on distribuido de prop´osito general [6] basado en el concepto de conjuntos de datos distribuidos resilientes (RDDs). Los RDDs son colecciones de elementos sobre los que se puede operar en paralelo en los nodos de un cl´uster, mediante el uso de dos tipos de operaciones: transformaciones (map, filter, union, etc.) y acciones (reduce, collect, count, etc.). Las caracter´ısticas principales de Spark son: modelo de programaci´on paralela de alto nivel, algoritmos de aprendizaje autom´atico, procesamiento de grafos, API de programaci´on multi-lenguaje, se ejecuta en diferentes sistemas (Hadoop, Mesos, Standalone, Cluster) y procesamiento de datos enstreaming. Spark se est´a creciendo en popularidad y ya est´a reemplazando a MapReduce (MR) como la tecnolog´ıa dominante para desarrollar aplicaciones Big Data. Como jMetalSP est´a orientado a Big Data Optimization, la parte que m´as interesa de Spark es su capacidad de procesamiento en paralelo de diferentes fuentes de datos en streaming, como: Kafka, Flumme, Twitter, sockets TCP, archivos, etc. Un Framework Basado en jMetal y Spark 5 Figura 2. Arquitectura de jMetalSP 3. Arquitectura de jMetalSP La architectura of jMetalSP se muestra en la Figura 2. Un aplicaci´on jMetalSP tiene como fin resolver un problema din´amico de optimizaci´on utilizando un algoritmo din´amico mediante el an´alisis de una o m´as fuentes de datos en streaming, existiendo uno o varios consumidores de datos que ir´a generando el algoritmo mientras se ejecuta. Como el problema se puede inicializar de varias maneras (con valores por defecto, a partir de datos externos, etc.) se utiliza un objeto que implementa el patr´on Builder. de forma similar, existe un builder para configurar y crear cada algoritmo. La clase SparkRuntime se usa para configurar la componente Spark de la aplicaci´on. El par´ametro m´as importante es el intervalo de procesamiento (batch interval). Como las fuentes de datos modificar´an el problema en tiempo de ejecuci´on, se debe definir un formato com´un que consiste en definir una subclase de la interfaz UpdatedData. Siempre que se modifica un problema, si se invoca sobre ´el el m´etodo isTheProblemModified(), ´este devolver´a el valor verdadero hasta que se invoque el m´etodo reset(). La clase jMetalSPApplication proporciona m´etodos para indicar qu´e builders del problema y algoritmo se van a usar, el Spark runtime, los consumidores de datos y las diferentes fuentes de datos. El m´etodo run() incorpora toda la l´ogica para crear el problema, el algoritmo, inicializar Spark, ejecutar los consumidores de datos y arrancar los componentes de procesamiento de los streams. Configurar y ejecutar una aplicaci´on jMetalSP consiste en crear e inicializar un objeto de jMetalSPApplication, tal como se muestra en el siguiente ejemplo: 6 Crist´obal Barba-Gonz´alez et al. 1. SparkSPApplicaton application = new SparkSPApplication() ; 2. application 3. .setSparkRuntime(new SparkRuntime(2)) 4. .setProblemBuilder(new SpecificProblemBuilder()) 5. .setAlgorithmBuilder(new SpecificAlgorithmBuilder()) 6. .addAlgorithmDataConsumer(new FirstDataConsumer()) 7. .addAlgorithmDataConsumer(new SecondDataConsumer()) 8. .addStreamingDataSource(new FirstStreamingSource()) 7. .addStreamingDataSource(new SecondStreamingSource()) 8. .addStreamingDataSource(new ThirdStreamingSource()) 9. .run() Cabe indicar que este c´odigo contiene algunas simplificaciones con el fin de hacerlo m´as f´acil de entender. El c´odigo real incluye gen´ericos de Java para garantizar en tiempo de compilaci´on que todos los componentes son compatibles. 4. Caso de Estudio: TSP din´amico con dos objetivos y NSGA-II din´amico Para evaluar la arquitectura propuesta se ha definido un caso de estudio que mezcla un problema acad´emico con datos reales. El problema din´amico seleccionado es el del viajante de comercio (Traveling Salesman Problem, TSP) con dos objetivos. Para resolverlo se usa una variante din´amica de NSGA-II. Los objetivos a minimizar son el tiempo y la distancia totales en recorrer todos los puntos de la instancia. Con el desarrollo de este caso de uso se busca un doble fin, primero mostrar c´omo se utiliza jMetalSP para implementar las versiones din´amicas, tanto del problema TSP como de la metaheur´ıstica NSGA-II; segundo, realizar pruebas para comprobar el rendimiento de la aplicaci´on en un cl´uster Hadoop/Spark. 4.1. NSGA-II din´amico La estrategia llevada a cabo para implementar una versi´on din´amica de NSGA-II usando la versi´on est´atica ya proporcionada por jMetal 5 se puede dividir en dos grandes pasos. En primer lugar, se buscan los m´etodos que necesitan ser modificados para permitir el comportamiento din´amico. En segundo lugar, se decide la configuraci´on del algoritmo. Con respecto al primer paso, se requiere modificar ´unicamente dos m´etodos: isStoppingConditionReached(): En el algoritmo est´andar de NSGA-II, cuando se alcanza el n´umero m´aximo de evaluaciones el algoritmo termina. Sin embargo, en la versi´on din´amica, se utilizan notificaciones para indicar que un nuevo frente de Pareto ha sido calculado. Despu´es de esto, el algoritmo se reinicia y comienza otra ejecuci´on. updateProgress(): Este m´etodo se utiliza para aumentar el n´umero de soluciones evaluadas en la versi´on est´atica de NSGA-II. En la versi´on din´amica, se modifica para comprobar tambi´en si el problema din´amico ha cambiado y, si es as´ı, se invoca el m´etodo reset() en el problema. Un Framework Basado en jMetal y Spark 7 Figura 3. Instancia del problema real TSP din´amico: 93 nodos (localizaciones geoposicionadas de calles) de la ciudad de Nueva York. La clase DynamicNSGAII resultante hereda de la clase original de NSGAII y sobreescribe estos dos m´etodos. Este enfoque de desarrollo se puede utilizar f´acilmente con la mayor´ıa de las metaheur´ısticas incluidas en jMetal 5. 4.2. TSP din´amico con dos objetivos Un problema din´amico en jMetalSP es un problema de jMetal 5 que implementa la interfaz DynamicProblem, por lo que incluye el m´etodo update(), que ser´a el que dar´a lugar a alg´un cambio en el problema. Dado que este m´etodo ser´a llamado por un objeto StreamingDataSource, el cambio resultante puede detectarse mediante otra clase (llamando a isTheProblemModified ()). Esto obliga el etiquetado como synchronized a todos los m´etodos del problema (incluyendo reset()). La clase DynamicMultiobjectiveTSP incluye los m´etodos antes descritos. Las variables de estado del problema son las matrices de distancia/coste. La instancia de TSP din´amico que se utiliza en este trabajo se fundamenta datos reales. M´as concretamente, se han usado los datos abiertos proporcionados por el Departamento de Transporte de la ciudad de Nueva York, el cual actualiza la informaci´on del tr´afico varias veces por minuto7. La informaci´on se proporciona mediante ficheros de texto donde cada l´ınea indica, entre otros datos, la velocidad media en pasar entre dos puntos de un intervalo. Despu´es de procesar los enlaces proporcionados por el servicio de tr´afico, el TSP din´amico resultante se compone de 93 ubicaciones y 315 caminos directos 7En la URL: http://207.251.86.229/nyc-links-cams/LinkSpeedQuery.txt 8 Crist´obal Barba-Gonz´alez et al. entre ellas, representadas en la Figura 3. Hay que se˜nalar que los enlaces son bidireccionales, por lo que el TSP resultante es asim´etrico. Los datos del problema son inicializados desde fichero por un objeto que implementa el interfaz AlgorithmBuilder. 4.3. Fuentes de datos en Streaming Para este trabajo, se han implementado tres clases de fuentes de datos en streaming:StreamingDirectoryTSP,StreamingKafka yStreamingTwitter. Los datos de tr´afico de Nueva York se leen peri´odicamente por una aplicaci´on externa que escribe en un directorio un archivo cada vez que se adquieren nuevos datos. La clase StreamingDirectoryTSP procesa los ficheros que se est´an agregando a ese directorio. Esta clase debe implementar el interfaz StreamingDataSource (ver Figura 1), que define el m´etodo start(). El aspecto m´as importante a tener en cuenta desde el punto de vista del rendimiento es que el tratamiento de los datos de entrada en streaming se realiza en paralelo, por lo que es en esta parte donde las caracter´ısticas del cl´uster de Spark ser´an m´as ´utiles. En nuestra clase StreamingDirectoryTSP, este procesamiento es muy simple, ya que el proceso externo actualiza el directorio dos o tres veces por minuto, por lo que no hay beneficios del uso del paralelismo. Por este motivo, hemos incluido otras dos fuentes de datos de secuencias (en las dos clases denominadas StreamingTwitterTSP yStreamingKafkaTSP) para profundizar en este tema. La clase StreamingTwitterTSP lee tweets de Twitter con el topic New York Traffic y el procesamiento de cada tweet es simulado, es decir, el problema se actualiza con un valor aleatorio. De esta manera se combina una fuente de flujo diferente con la posibilidad de ajustar el tiempo de procesamiento. Por ´ultimo, la clase StreamingKafkaTSP est´a destinada, como la anterior, a enriquecer el caso de estudio con otra fuente de datos. Para este caso, hemos creado un productor de mensajes Kafka que genera, siguiendo distribuciones uniformes y normales, una serie de mensajes aleatorios con datos para actualizar el problema. Cada cinco segundos se producen al menos 1.000 mensajes, aunque en promedio se crean alrededor de 10.000 mensajes. StreamingKafkaTSP lee los mensajes creados por este productor y actualiza el problema cada segundo. 4.4. Experimentos Para la evaluaci´on de la arquitectura se han realizado tres experimentos. Para ello se ha usado un sistema virtualizado con VMWare con 10 m´aquinas esclavas, cada una con 10 cores a 2.7 Ghz, 10 GB RAM y 100 GB de disco duro. Adem´as se dispone de una maquina f´ısica usada como maestra con 8 procesadores i7 a 3.40 GHz, 32 GB RAM y 3 TB de disco duro. La diferencia entre cada uno de los test realizados es la cantidad y forma en el que se crean los datos. Se disponen de tres fuentes de datos, mensajes Kafka, ficheros con datos reales de tr´afico de Nueva York y tweets sobre el tr´afico en Nueva York. Para aumentar el n´umero de datos usado en el problema y comprobar la eficiencia de la lectura en streaming, Un Framework Basado en jMetal y Spark 9 Figura 4. Experimento 3. Uso de memoria del sistema (izquierda) y n´umero de thread/minuto creados en cada m´aquina (derecha) se usa el productor de mensajes de Kafka para variar en la cantidad de datos que se deben procesar en el sistema. Para los experimentos 1 y 2 se producen los mensajes de Kafka siguiendo una distribuci´on aleatoria uniforme llegando a crear cada 5 segundos una media de 50.000 mensajes (140 GBs de datos le´ıdos durante la prueba) para el experimento 1 y de 500.000 (1.37 TBs de datos le´ıdos durante la prueba) para el experimento 2. Este segundo experimento lleva a un colapso del sistema debido a que no es capaz de leer todos los datos generados. Por otro lado, para el experimento 3 se producen tambi´en 500.000 mensajes cada 5 segundos (1.37 TBs de datos le´ıdos durante la prueba) pero siguiendo una distribuci´on normal, lo que favorece a una mejor distribuci´on de la lectura de los datos en streaming, evitando as´ı el colapso del sistema. En la Figura 4 se puede observar la cantidad de memoria usada durante la ejecuci´on del experimento 3 (1 hora) junto con el n´umero de threads por minuto creados (y aceptados por CPUs) en cada una de los nodos del cl´uster. Como se puede observar, se realiza un uso intensivo de memoria y CPU, aunque sin sobrepasar los l´ımites virtuales de la plataforma debido al sobrecoste en operaciones de swap ybuffering. 5. Conclusiones y Trabajo Futuro En el presente trabajo se ha presentado jMetalSP, una soluci´on software para hacer frente a problemas de optimizaci´on din´amica en entornos Big Data, que combinan el framework de optimizaci´on jMetal con el sistema de cluster computing Apache Spark. La motivaci´on de este trabajo ha sido impulsada por el aumento en el uso de Spark como plataforma de computaci´on distribuida, as´ı como la utilizaci´on del framework jMetal como motor de optimizaci´on multi-objetivo. Se ha generado un caso de estudio que considera una formulaci´on del problema TSP con dos objetivos: minimizar la distancia total y el tiempo de viaje. Se ha generado una instancia real del problema creada a partir de los datos abiertos de la ciudad de Nueva York. Dichos datos se han usado para actualizar la informaci´on del problema durante la ejecuci´on. Adem´as, se han tratado dos fuentes