Administración eficiente derecursos en MapReduce

Tesis de Ingeniería en Informática

Federico Retyk

Director:Dr. Juan M. Ale

Facultad de IngenieríaUniversidad de Buenos Aires

16 de junio de 2015

Resumen

El componente de planificación es crítico para la eficiencia en el procesa-miento de datos bajo el paradigma MapReduce. Es frecuente que los clustersque implementan Hadoop se utilicen en forma compartida entre múltiplesusuarios, con distintos niveles de carga de trabajo. Cuando dos o más jobsse ejecutan en simultáneo, estos compiten por los recursos del cluster, por locual la ausencia de un mecanismo inteligente de asignación de recursos po-dría afectar seriamente la performance del sistema. Se elabora en la presentetesis un exhaustivo análisis sobre la administración eficiente de recursos enMapReduce y se propone un nuevo esquema de planificación, basado en elestado del arte del tema. A partir de resultados experimentales basados ensimulación, se establece que el componente de scheduling propuesto, denomi-nado Flexible Coupling Scheduler, logra mejorar el tiempo promedio deprocesamiento de los jobs en hasta un 33,7% respecto a la implementaciónestándar, Fair Scheduler.

Palabras clave: Digitalización, Big Data, Data Analytics, Minería dedatos, Aprendizaje automático, Computación concurrente, MapReduce, Ha-doop Coupling Scheduler, Asignación de tareas Reduce

i

Abstract

The scheduling component proves to be critical for efficient proccessing un-der the MapReduce paradigm. Hadoop clusters are usually shared amongmultiple users, with different work-load levels. When two or more jobs areexecuted simultaneously, they compete for the cluster resources, for whichthe absense of an intelligent schedule algorithm can seriusly impact systemperformance. In this thesis, we present a thorough analysis on efficient re-source management in MapReduce and propose a novel scheduler, namelyFlexible Coupling Scheduler, based on state-of-the-art research. Oursimulated experiments showed improvements to flowtime and job responsetimes by up to 33,7% compared with Fair Scheduler.

Keywords: Digitaization, Big Data, Data Analytics, Data Mining, Ma-chine Learning, Parallel Computing, MapReduce, Hadoop Coupling Sched-uler, ReduceTask Scheduling

ii

Índice general

1. Introducción 11.1. Motivación y objetivos . . . . . . . . . . . . . . . . . . . . . . 21.2. Descripción de la contribución . . . . . . . . . . . . . . . . . . 31.3. Organización del documento . . . . . . . . . . . . . . . . . . . 4

2. Contexto 52.1. Digitalización . . . . . . . . . . . . . . . . . . . . . . . . . . . 5

2.1.1. Gestión de datos . . . . . . . . . . . . . . . . . . . . . 62.1.2. Extracción del conocimiento . . . . . . . . . . . . . . . 7

2.2. Desafíos de Big Data . . . . . . . . . . . . . . . . . . . . . . . 102.2.1. Computación concurrente . . . . . . . . . . . . . . . . 102.2.2. Aprendizaje automático . . . . . . . . . . . . . . . . . 122.2.3. Visualización . . . . . . . . . . . . . . . . . . . . . . . 14

2.3. MapReduce . . . . . . . . . . . . . . . . . . . . . . . . . . . . 172.3.1. Definición del modelo . . . . . . . . . . . . . . . . . . 192.3.2. Plataforma subyacente . . . . . . . . . . . . . . . . . . 202.3.3. Comparación con otros modelos y sistemas . . . . . . 21

2.4. Hadoop . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 232.4.1. Hadoop Distributed File System . . . . . . . . . . . . 242.4.2. MapReduce runtime . . . . . . . . . . . . . . . . . . . 272.4.3. Scheduler . . . . . . . . . . . . . . . . . . . . . . . . . 292.4.4. YARN . . . . . . . . . . . . . . . . . . . . . . . . . . . 33

2.5. Spark . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 352.5.1. Resilient Distributed Datasets . . . . . . . . . . . . . . 372.5.2. Operaciones soportadas . . . . . . . . . . . . . . . . . 392.5.3. Persistencia en memoria de acceso aleatorio . . . . . . 402.5.4. Scheduling y comunicación de datos . . . . . . . . . . 412.5.5. Casos de uso de Spark y Hadoop/MapReduce . . . . . 43

iii

3. Estado del arte 463.1. Implementaciones del Scheduler de Hadoop . . . . . . . . . . 47

3.1.1. Fair Scheduler . . . . . . . . . . . . . . . . . . . . . . . 473.1.2. Capacity Scheduler . . . . . . . . . . . . . . . . . . . . 493.1.3. LATE Scheduler . . . . . . . . . . . . . . . . . . . . . 503.1.4. Resource-aware Adaptive Scheduler . . . . . . . . . . . 523.1.5. COSHH . . . . . . . . . . . . . . . . . . . . . . . . . . 533.1.6. Dynamic Priority Scheduler . . . . . . . . . . . . . . . 543.1.7. Coupling Scheduler . . . . . . . . . . . . . . . . . . . . 55

3.2. Optimizaciones en la asignación de ReduceTasks . . . . . . . 633.2.1. Enfoques greedy . . . . . . . . . . . . . . . . . . . . . . 643.2.2. Asignación secuencial estocástica . . . . . . . . . . . . 713.2.3. Heurística RHCP . . . . . . . . . . . . . . . . . . . . . 753.2.4. Algoritmo SRT . . . . . . . . . . . . . . . . . . . . . . 77

3.3. Trabajos relacionados . . . . . . . . . . . . . . . . . . . . . . 78

4. Contribución 814.1. Modelización del problema . . . . . . . . . . . . . . . . . . . . 81

4.1.1. Notación y definición de conceptos . . . . . . . . . . . 824.1.2. Determinación del problema . . . . . . . . . . . . . . . 86

4.2. Solución propuesta . . . . . . . . . . . . . . . . . . . . . . . . 874.2.1. Descripción del algoritmo . . . . . . . . . . . . . . . . 884.2.2. Diseño de la implementación . . . . . . . . . . . . . . 89

4.3. Discusión . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 94

5. Resultados experimentales 975.1. Métricas a tener en cuenta . . . . . . . . . . . . . . . . . . . . 975.2. Experimentación . . . . . . . . . . . . . . . . . . . . . . . . . 99

5.2.1. Metodología . . . . . . . . . . . . . . . . . . . . . . . . 995.2.2. Diseño . . . . . . . . . . . . . . . . . . . . . . . . . . . 1015.2.3. Resultados . . . . . . . . . . . . . . . . . . . . . . . . 103

5.3. Análisis de los resultados . . . . . . . . . . . . . . . . . . . . . 115

6. Conclusiones y trabajo futuro 1186.1. Conclusiones . . . . . . . . . . . . . . . . . . . . . . . . . . . 1196.2. Futuras líneas de investigación . . . . . . . . . . . . . . . . . 119

A. Nomenclatura 121

B. Código fuente 125

iv

1

Introducción

En la actualidad existe un amplio consenso en el hecho de que lograradquirir información valiosa a partir de los datos disponibles es un asun-to crítico y de enorme potencial. Perfeccionar el proceso de obtención deconocimiento empírico encierra la posibilidad de optimizar el desarrollo deinvestigaciones científicas y mejorar la toma de decisiones en entornos denegocios.

Los avances tecnológicos en el área de informática y telecomunicacionesfacilitaron la recolección y almacenamiento de datos, dando lugar a un ritmode crecimiento de los datos almacenados que, hoy se estima, es a duplicarsecada año [1]. En tanto que esta expansión formidable del universo de datosdisponibles (es decir, la totalidad de datos creados y almacenados actualmen-te) es incluso mayor que las mejoras en tiempo de acceso y procesamiento,y dada la complejidad que existe para trabajar en paralelo de forma efi-ciente, resulta ser un desafío la explotación de aquellos datos para extraerconocimiento. Por otro lado, dicha expansión también brinda grandes opor-tunidades para los procesos de análisis, dado que alimentarlos con una mayorcantidad de datos permite obtener resultados más certeros.

MapReduce es un modelo de programación que tiene por objetivo au-tomatizar los aspectos de procesamiento paralelo de datos, permitiendo alos desarrolladores enfocarse en las cuestiones de alto nivel de las tareas deanálisis de datos [2]. Hadoop, entre otras implementaciones de MapReduce,se aplica en la actualidad a una gran variedad de problemas científicos eindustriales, en los que la capacidad para procesar datos a gran escala y entiempos razonables es un factor crítico.

1

1.1. Motivación y objetivos

El beneficio que proporcionan las tecnologías de procesamiento eficiente ymasivo de datos al ayudar a comprender problemas complejos de forma cer-tera y profunda, para lo cual generalmente hace falta manipular importantescantidades de información, es el principal motivo para elegirlas como temaa desarrollar en este trabajo. A continuación se describen algunos ejemplosnotables de problemas abordados mediante este tipo de tecnologías.

Entre las aplicaciones científicas de este tipo de tecnologías para resol-ver problemas de medicina, el caso más paradigmático es quizás el de lasecuenciación del ADN humano, que busca capturar digitalmente y explotarla información genética de las personas para estudiar sus procesos biológi-cos fundamentales, diagnosticar enfermedades y descubrir o perfeccionar sustratamientos [3]. Puesto que dicho proceso genera enormes cantidades dedatos, y dado que los sistemas tradicionales de procesamiento de datos nopueden hacerle frente, surgieron herramientas que utilizan el modelo MapRe-duce para analizar las secuencias y extraer conocimiento, como es el caso delframework GATK [4].

Si bien los costos de secuenciación son elevados e impiden su aplicaciónsistemática en el tratamiento de pacientes, la tendencia registrada durantelos últimos años es de reducción dramática de estos costos [5]. En caso deque efectivamente disminuyan los costos de secuenciación, esto abriría laspuertas a variadas oportunidades interesantes de atención personalizada delos pacientes, en tanto que sería posible analizar la secuencia del pacientecomo un vector y buscar asociaciones estadísticas en la base de datos dehistorial clínico para encontrar casos similares registrados, y poder predecirde forma más certera la respuesta ante distintos cursos de tratamiento paraoptimizar los resultados y evitar posibles efectos adversos.

Otro caso interesante es el de las denominadas smart cities. Por ejemplo,entre las diversas iniciativas tecnológicas que impulsó el Ajuntament de Bar-celona para mejorar los servicios públicos de la ciudad, se destaca el proyectopuesto en marcha durante la Mercè [6] en el año 2012. En este caso se utilizóinformación disponible en las redes sociales (con la ubicación, fotografías ymensajes que incluyan descripciones de las experiencias del público) paraoptimizar los servicios públicos que dependen del Ajuntament, como la or-ganización del sistema de transporte, las fuerzas de seguridad y los serviciosde emergencias.

2

El gobierno municipal de Copenhage propone también procesar los datosdisponibles con el fin de optimizar el uso de energía y desarrollar solucionesnuevas que ayuden a mejorar los hábitos del uso de energía. Entre otrosbeneficios del enfoque, se alcanzó actualmente una reducción del 40% en lasemisiones de CO2 a la atmósfera, y se espera reducirla completamente parael año 2025 [7].

Del mismo modo, las organizaciones pueden obtener información útil ypotencialmente valiosa mediante la minería de sus logs de datos. En Twit-ter, por ejemplo, se utiliza Hadoop para llevar adelante tareas de minería yaprendizaje automático a gran escala [8], mediante un sistema denominadoElephant Bird (que resulta de un juego de palabras con los logotipos de lacompañía y la plataforma de procesamiento). Así, procesar la informacióncontenida en la red social de microblogging les permite detectar spam y pre-decir intereses de los usuarios para realizar sugerencias que estos acepten conmayor probabilidad, entre otras funcionalidades.

Los objetivos del presente trabajo son proporcionar un análisis extensosobre la administración eficiente de recursos en el procesamiento rápido ymasivo de datos bajo el modelo de paralelización automática MapReduce (yparticularmente en Hadoop, realización open-source de dicho modelo), y eldiseño e implementación de un Scheduler optimizado, basado en las mejorasmás recientes.

1.2. Descripción de la contribución

La contribución de la tesis, orientada a cumplir los objetivos descriptosen la sección previa, está compuesta por los siguientes aportes:

1. Un análisis detallado sobre el uso de recursos en el modelo MapReducey la plataforma distribuida de código abierto Hadoop.

2. El estudio del estado del arte en materia de administración de recursosen Hadoop.

3. La modelización formal del problema de optimización de la función deasignación de tareas Reduce por parte del componente de schedulingde la plataforma mencionada.

4. Una solución propuesta para dicho problema de optimización, basadoen las mejoras más recientes, y evaluada mediante simulación.

3

1.3. Organización del documento

A continuación, el documento se encuentra estructurado de la siguientemanera. El Capítulo 2 presenta el contexto del trabajo, en tanto que describela temática de la digitalización de información, los desafíos que surgen delcrecimiento en la cantidad de datos almacenados y las tecnologías utilizadaspara abordar su procesamiento. En el Capítulo 3 se detalla el estado delarte de dichas tecnologías. Luego se desarrolla la contribución de la tesisen el Capítulo 4, y los resultados de la experimentación llevada adelantese exponen en el Capítulo 5. Finalmente, en el Capítulo 6 se elaboran lasconclusiones del trabajo, y se proponen temas para futuras investigaciones.

4

2

Contexto

El presente capítulo aborda en general el tema de la digitalización de lainformación y el potencial que encierran las herramientas que existen pa-ra analizar los datos y extraer conocimiento. Asimismo, con el objetivo depresentar un contexto para el trabajo posterior de la tesis, se describe lasituación actual de crecimiento extraordinario de las cantidades de datosalmacenados en los sistemas de información, así como los problemas y lasoportunidades que esto genera.

Se analizarán los conceptos relacionados y se tratarán algunos de los prin-cipales componentes de MapReduce, un modelo de procesamiento masivo dedatos ampliamente utilizado en proyectos científicos e industriales. Dicho mo-delo cuenta en la actualidad con una comunidad de investigación y desarrolloen continuo crecimiento, especialmente en relación a Hadoop, su implemen-tación open-source de mayor popularidad, y Spark, sistema surgido comoextensión de MapReduce para obtener mayor generalización y optimizacióndel uso de la memoria.

2.1. Digitalización

Capturar datos, surgidos a partir de cualquier clase de fenómeno de in-terés, para almacenarlos en sistemas digitales es el proceso conocido comodigitalización. Este proceso, potenciado en la actualidad por el desarrolloexperimentado por las tecnologías de la información, permite aprovechar almáximo los datos que se tienen a disposición sobre el fenómeno en estudio

5

para generar conocimiento útil y valioso.

Ya sea con fines académicos o de negocios, el análisis sistemático de losdatos capturados puede poner al descubierto respuestas a interrogantes cien-tíficos e industriales, y posibilita extraer conocimiento estratégico que res-palde la toma de decisiones para obtener mejores resultados. A través delprocesamiento continuo e iterativo de la información empírica disponible (ymediante el testeo de hipótesis o productos bajo análisis), se logra refinar elentendimiento que se tiene del fenómeno [9].

Estas plataformas de información [10], que motorizan la captura, análisisy explotación de los datos recabados, deben poseer la capacidad de incorporartodos los datos disponibles que puedan resultar de interés, deben ser lo sufi-cientemente flexibles como para adaptarse ante la evolución de los esquemasy modelos, y deben permitir ejecutar operaciones complejas sobre aquellosdatos. De esta forma pueden optimizar y acelerar el proceso de extraccióndel conocimiento.

Desde el punto de vista de la ingeniería de las plataformas de informa-ción, resulta necesario diseñar, construir y administrar estructuras capacesde almacenar y procesar de forma eficiente los crecientes volúmenes de datoscapturados. A partir de allí, el uso de dichos sistemas para explotar y analizaresos datos aplicando conocimientos de estadística y aprendizaje automático,componen los aspectos científicos del problema.

2.1.1. Gestión de datos

Si bien el estudio de la gestión de los datos es un campo amplio e intere-sante (en la publicación de 2009 de DAMA International [11] se encuentraextensamente desarrollado), a continuación se hará mención de aquellos con-ceptos relacionados al tema del presente trabajo.

En la actualidad, un sistema típico de business intelligence consta de unconjunto de herramientas: en primer lugar, un ETL (siglas del término en in-glés para extracción, transformación y carga) que toma periódicamente datosde una colección de fuentes, y las prepara con el fin de permitirlas almace-nar con las estructuras disponibles del sistema; un Data Warehouse [12, 13],que funciona como estructura de almacenamiento de los datos extraídos porel ETL; y herramientas de análisis que procesan los datos disponibles en elwarehouse para generar reportes y estadísticas para consumo interno de laorganización.

6

El diseño de los Data Warehouses está orientado a poner a disposición deforma rápida y eficiente la información almacenada. Para ello, los sistemastradicionales almacenan la información de manera estructurada y utilizandobases de datos relacionales. Si bien esto supera en términos de tiempo deacceso a otros tipos de plataformas de información, como aquellos denomi-nados NoSQL, también suponen una mayor dificultad ante el crecimientode la información, puesto que su implementación en sistemas distribuidos esextremadamente compleja. Otro problema de este tipo de sistemas de alma-cenamiento es su falta de flexibilidad para trabajar de forma performantecon datos no-estructurados, que son abundantes en la red y de los cualestambién podría extraerse conocimiento valioso.

El término NoSQL (o más puntualmente, Not-Only-SQL) se encuentratodavía sin una definición precisa. No obstante, hace referencia a bases dedatos no-relacionales como BigTable [14], Dynamo [15], Cassandra [16] yMongoDB [17, 18], que manipulan datos que no responden a esquemas es-pecíficos y trabajan sobre clusters de computadoras. De esta forma, tienenla habilidad de sacrificar en cierta medida la consistencia de los sistemastradicionales para obtener mejoras en otros aspectos como performance, es-calabilidad y facilidad de uso [19].

Este nuevo tipo de sistemas de almacenamiento de datos no reempla-zan a los sistemas tradicionales, sino que son complementarios. Las bases dedatos relacionales son herramientas poderosas y tienen, para ciertas aplica-ciones, mejores resultados que aquellas NoSQL. Pero en casos en los cualesse requiere manipular grandes cantidades de datos (donde entran en juegolos clusters), trabajar con grafos o con información que proviene de muchasfuentes heterogéneas, las bases de datos no-relacionales pueden resultar másadecuadas.

2.1.2. Extracción del conocimiento

Bajo el término Knowledge Discovery in Databases (KDD) se conoce a ladisciplina que estudia el conjunto de procedimientos y recursos para identifi-car, en los datos, patrones significativos que sean válidos, novedosos, poten-cialmente útiles y comprensibles para un usuario [20]. Esta disciplina se valede herramientas que provienen de diversos campos de investigación: estadís-tica, inteligencia artificial, teoría de bases de datos, computación distribuiday teoría de visualización de datos, entre otros.

7

Si bien fue inicialmente acuñado por Piatetsky-Shapiro para poner enevidencia que el objetivo final del proceso es obtener conocimiento a partirde un set de datos [21], el término extendió posteriormente su significadopara abarcar y definir en detalle todas las fases que componen el procesode extracción del conocimiento, ilustradas en la figura 2.1. Estas etapas se-rán brevemente descriptas a continuación (en la sección de bibliografía seincluyen referencias a trabajos que ofrecen información detallada de dichoproceso [22, 23], así como también algunos ejemplos interesantes de aplica-ción en distintas áreas académicas e industriales [24]).

La primera fase del proceso de KDD consiste en seleccionar el conjuntode datos iniciales, que formarán parte de la entrada del proceso. Esto suelerequerir la integración de distintas bases de datos, y la discriminación en-tre fuentes de información que proporcionen mejores resultados del proceso,respecto de aquellas que carecen de valor y que no aportan o incluso puedanperjudicar.

Asimismo, puesto que la calidad del conocimiento extraído por el procesode KDD depende considerablemente de la calidad de los datos estudiados,se incluyen las etapas de preprocesamiento y transformación de los datos deentrada. Su importancia entonces radica en preparar los datos de entradade la mejor manera posible para que, alimentando la subsiguiente etapa deminería, se obtengan luego mejores resultados.

Como ejemplo de la fase de transformación de variables, Jolliffe [25] ofreceuna presentación sobre el método de componentes principales, mientras quela fase de preprocesamiento se encuentra abordada en detalle en el segundocapítulo de la publicación de Han y Kamber [26].

Figura 2.1: Panorama de las fases involucradas en el proceso KDD [23].

8

La fase principal, o núcleo, del proceso consiste en la aplicación de algo-ritmos de inducción que, tras explorar los datos (ya preparados en las etapasanteriores), generan modelos matemáticos que capturan patrones significati-vos [22, 27] que son, en esencia, conocimiento potencialmente valioso. Paraello puede requerirse ejecutar sucesivas veces el algoritmo, hasta encontraruna configuración adecuada que lleve a la generación de un modelo satisfac-torio.

De acuerdo a la taxionomía presentada por Oded Maimon et al. [22],los métodos de minería de datos se clasifican en dos grandes grupos. Losmétodos de verificación ejecutan el proceso con el fin de evaluar una hipótesispropuesta por una fuente externa, como podría ser un experto del fenómenoen estudio, como podría ser un test de bondad de ajuste, análisis de varianzao un test de hipótesis, entre otros.

Los métodos orientados al descubrimiento de patrones previamente des-conocidos se dividen a su vez en dos sub-grupos. Los métodos descriptivosbuscan encontrar la relación que existe entre los elementos de los datos deentrada (por ejemplo, los métodos de clustering buscan agrupar los ejemen-tos de acuerdo a algún criterio arbitrario de similitud). Por el contrario, losmétodos de predicción buscan construír automáticamente modelos de com-portamiento, que permitan predecir el valor de una o más variables a partirde elementos desconocidos de la población de datos inicial.

Este último grupo se divide también entre modelos de regresión y modelosde clasificación. En el caso de los métodos de regresión, se tiene una ecuaciónpredeterminada, y el modelo se genera a partir de encontrar los parámetrosde dicha ecuación que mejor ajusten a los datos de prueba. Los métodos declasificación, por el contrario, buscan predecir el valor de una clase (variablediscreta) a partir del valor de las demás variables. Ejemplos de métodos declasificación son las redes neuronales, las redes bayesianas y los árboles dedecisión, entre otros.

Luego, durante la etapa final del proceso de extracción de conocimiento,se interpretan los patrones obtenidos en la etapa anterior, regresando si esnecesario para repetir cualquiera de los pasos anteriores. Para ello es fre-cuente la aplicación de métodos de visualización, que permiten interpretarlos datos de entrada a la luz de los patrones y modelos surgidos de la fasede minería [27]. El gráfico 2.2 muestra algunos ejemplos de visualización,generados mediante la librería D3.js [28].

9

Figura 2.2: Visualizaciones interactivas producidas mediante D3.js [28].

2.2. Desafíos de Big Data

Big Data es un término que hace referencia al crecimiento exponencial ya la disponibilidad de los datos, tanto estructurados como desestructurados,y a los sistemas de computación capaces de procesar de forma eficiente lainformación en estas circunstancias.

Si bien el concepto carece de una definición rigurosa, en la actualidad seevalúa si un problema pertenece a esta categoría de acuerdo a la definiciónde Doug Laney [29], considerando tres métricas relativas a los datos: el vo-lumen o la cantidad de datos disponibles, la velocidad con la cual crecen yse modifican, y la variedad de los tipos y fuentes de donde se obtienen.

La necesidad de extraer conocimiento a partir de cantidades masivas dedatos presenta el desafío de desarrollar herramientas y algoritmos que dispon-gan de los recursos, de forma eficiente, para llevar adelante el procesamientode los datos y obtener resultados satisfactorios en tiempos razonables.

Además, ante la enorme complejidad de la información extraída, se tor-na indispensable encontrar formas claras de visualizar los resultados paraoptimizar el proceso cognitivo.

2.2.1. Computación concurrente

La figura 2.3 da cuenta que en los últimos años dejó de registrarse larelación descripta por la famosa observación de Gordon E. Moore, quiensugería que la cantidad de transistores que contienen los circuitos integradosse duplica aproximadamente cada dos años.

10

Figura 2.3: Evolución de la performance de los procesadores [30].

Esto se debe en parte a que se están alcanzando límites físicos para cons-truír capacitores más pequeños, que actualmente son del orden del tamañoatómico, y en parte también a que el ancho de banda de memoria no evolu-cionó tan favorablemente, con lo cual no tiene sentido construír procesadoresmás rápidos (ya que quedarían constantemente a la espera de la memoriaque se copia desde caché) [30, 31].

Es por ello que en la actualidad, la tendencia es a construír sistemasdistribuidos de muchas computadoras trabajando en concurrencia, en lugarde buscar construír computadoras más rápidas1.

Dado el alto nivel de complejidad de los sistemas informáticos, en tér-minos de la cantidad de componentes y la complejidad de la interacciónentre ellos, resulta indispensable para su análisis (y para el diseño de dichossistemas) contar con abstracciones.

En programación, la abstracción del concepto de concurrencia implica elestudio de secuencias superpuestas que ejecutan las instrucciones atómicasde un proceso secuencial [32]. Por el contrario, un proceso secuencial refiere ala ejecución sucesiva de instrucciones atómicas, donde cada ejecución precedea otra sin que nunca dos de ellas sean simultáneas.

1En su célebre cita, Grace Hopper fundamenta la necesidad de trabajar con sistemas demás computadoras utilizando como analogía el empleo de una mayor cantidad de bueyespara tirar del arado, en lugar de buscar hacerlos crecer más de lo posible.

11

Un programa concurrente, entonces, está compuesto por la ejecución po-siblemente simultánea de un conjunto de procesos secuenciales. Este tipo deprogramas permiten acelerar ciertas tareas, compuestas por pasos que pue-den ser ejecutados de forma independiente, al dividirlas para procesar dichospasos paralelamente en distintos flujos de procesos, acortando así el tiempototal requerido para la tarea. Para ello es necesario contar con sistemas quepermitan efectuar esta paralelización.

Un sistema distribuido es aquel en el cual los componentes de hardwareo software de una red de computadoras se comunican entre sí y coordinansus operaciones mediante el envío de mensajes [33]. Este tipo de sistemasdeben contar con un buen nivel de escalabilidad, es decir, la propiedad quedetermina en qué medida continúan siendo eficientes ante el incremento derecursos y usuarios. Además, requieren de protocolos de comunicación queposibiliten el envío de mensajes entre equipos heterogéneos (esto es logradomediante estándares como el suite de protocolos TCP/IP [34], entre otros).

Asimismo, al tratarse de sistemas de múltiples componentes, existe la po-sibilidad de que algunos de ellos fallen mientras otros continúan funcionando.Por lo tanto, para lograr un nivel aceptable de robustez, resulta indispensableproveerlos con mecanismos de gestión de fallas que permitan detectarlas ytomar medidas correctivas para solucionarlas o al menos reducir su impacto.

2.2.2. Aprendizaje automático

Las herramientas de aprendizaje automático (machine learning) resultanmuy valiosas, si no indispensables, para afrontar los desafíos que imponetrabajar con cantidades masivas de datos. A su vez, las plataformas de pro-cesamiento de Big Data permiten aplicar métodos de aprendizaje automáticoen tiempos razonables sobre cantidades antes prohibitivas de datos.

Un ejemplo de aplicación interesante de aprendizaje automático es la quese denomina sentiment analysis (también, opinion mining), donde se hace usode la información disponible en las redes sociales para observar la opinión delas personas sobre ciertos temas [35].

Asimismo, el ranqueo de las páginas web que devuelve una consulta rea-lizada sobre un motor de búsqueda es otro problema donde se aplican herra-mientas de procesamiento masivo de datos para potenciar los algoritmos deaprendizaje automático [36].

12

En general, la información en las sociedades modernas se encuentra enconstante crecimiento y es almacenada en vastos repositorios de datos, tantopúblicos como privados, por lo cual los sistemas que aplican aprendizajeautomático deben poder escalarse acorde [37]. Se tiene entonces que existeuna retroalimentación significativa entre ambas disciplinas.

A continuación se introdce el tema de forma breve, si bien algunas obrasde referencia [38, 39, 40] pueden ser encontradas en la sección de bibliografía.

Sean X el espacio de entrada, Y el espacio de salida y un conjunto demuestras de entrenamientoD = (x1, y1), (x2, y2), · · · , (xn, yn) en el espacioX × Y (llamadas instancias). Entonces el problema general de aprendizajeautomático supervisado es encontrar la función f : X → Y que mejor seajuste al conjunto de entrenamiento, en términos de minimizar una funciónde error L que cuantifica la diferencia entre el valor f(xi) predicho por lafunción, y el valor real yi [41].

Una vez generado el modelo, i.e., fue seleccionada la mejor f del espaciode hipótesis, es posible utilizarlo para realizar predicciones sobre muestrasde dato cuyo valor asociado no se conoce (esto es, análisis predictivo). Por lotanto, una solución de aprendizaje automático consiste en tres componentes:los datos, los valores derivados de los mismos y el modelo obtenido.

Tal como señalan Lin et al. [8], sucede que la experiencia acumuladadurante los últimos años en aplicaciones del mundo real refuerzan la idea deque el tamaño de los datos es el factor más importante a la hora de obtenerresultados satisfactorios [42].

Además, existe una gran cantidad de investigaciones que sostienen quemodelos simples, entrenados sobre enormes cantidades de datos, obtienenmejores resultados que modelos más sofisticados, entrenados sobre conju-tos más pequeños de datos [43]. Esto dio lugar al crecimiento de solucionessimples, pero con el foco puesto en los datos [44, 45].

Uno de los métodos de aprendizaje automático más utilizado a la horade procesar datos a gran escala es Stochastic Gradient Boosting [46], y enparticular Stochastic GBDT (Gradient Boosting Decision Trees). En su pu-blicación de 2009, Jerry Ye et al. [47] proponen una versión distribuida delalgoritmo GBDT estocástico, que prueban ejecutando sobre Hadoop utili-zando primero una versión adaptada a MapReduce, y luego otra basada enMPI (Message Passing Interface).

13

2.2.3. Visualización

Los métodos y herramientas de visualización son críticos cuando se tra-baja con cantidades masivas de datos. Los gráficos o figuras deben reducirla complejidad del fenómeno en estudio, para ayudar a las personas que loutilizan a encontrar patrones en los datos.

Para optimizar la percepción de la información derivada de los datos,estas representaciones visuales deben ser legibles, claras y, en lo posible,atractivos.

La importancia de obtener visualizaciones efectivas de los datos radicaen el hecho de que para la percepción humana resulta más fácil y rápidoreconocer patrones a partir de gráficos o figuras, que a partir de texto, ecua-ciones o matrices. De esta forma es posible optimizar la presentación de lainformación generada en el proceso descripto en la Subsección 2.1.2.

Además, dado que el análisis de los datos por parte de especialistas resultaser otro paso en el ciclo de extracción de la información, la visualización delos mismos logra un trabajo más efectivo.

Existen herramientas como Circos [48] y D3.js [28] que facilitan la crea-ción de gráficos efectivos a partir de un set de datos de entrada. Los ejemplosexpuestos en la figura 2.2 son, de izquierda a derecha: una vista de calendario,un diagrama de cuerdas, un mapa coroplético, una agrupación jerárquica debordes, una matriz de diagramas de dispersión, un diagrama de barras agru-padas y ampliadas, un grafo basado en fuerzas clusterizado y un diagramade Voronoi.

Figura 2.4: Relación entre distintas secuencias de aminoácidos [49].

14

Los diagramas de cuerdas (chord diagrams) presentan información me-diante dos conjuntos de elementos: (a) un conjunto de arcos, denominadosgrupos, que determinan el peso o valor relativo del grupo respecto a los de-más, y figuran en el exterior del círculo, y (b) un conjunto de cuerdas, en elinterior del círculo, que establecen la intensidad de las relaciones entre losdistintos grupos de acuerdo a su grosor.

Este tipo de diagramas son útiles cuando se busca visualizar la relaciónentre grupos de entidades, y son comunmente utilizados, entre otras áreas,por la comunidad científica para representar datos genéticos, como puedeverse en la figura 2.4.

Los denominados hive plots son representaciones visuales de redes o gra-fos, que resulta de gran utilidad al tratar con datos masivos. Una represen-tación tradicional de grafo, cuando se cuenta con cantidades inmanejablesde nodos, adquiere generalmente una forma que hace dificil extraer patroneso visualizar información. En cambio, los hive plots distribuyen los nodos enejes que nacen desde un origen, y que pueden estar también divididos ensegmentos.

Se establecen reglas (o métricas) que determinan la posición de los nodos,por lo cual adquieren mayor o menor centralidad de acuerdo a las mismas.Además, los enlaces entre nodos se dibujan como curvas de Bézier, y tambiénpueden utilizarse colores o etiquetas para mejorar su nivel de claridad.

La figura 2.5 expone una estimación, elaborada por el equipo de NikolaSander et al. [50], de los flujos migratorios entre distintas regiones. Se tieneen este caso una visualización muy bien lograda (por su claridad y eficienciapara comunicar la información) de las cantidades relativas de inmigrantesdesde cada una de las regiones hacia las demás.

Otra clase de visualizaciones son los mapas coropléticos (choropleth map),donde se colorean las regiones de un mapa geográfico de acuerdo a metricasestablecidas sobre una variable estadística, como podrían ser la densidad depoblación o su nivel adquisitivo, la preferencia electoral de la población endistintas localidades, o cualquier otro tipo de información geográfica.

Un ejemplo interesante de visualización es el que aparece en la figura 2.6,que integra un trabajo de Timm Kekeritz2 donde se grafican los niveles detemperatura y precipitación en las grandes ciudades (en este caso, BuenosAires) durante todos los días del año 2013.

2http://www.weather-radials.com/

15

Figura 2.5: Flujos migratorios entre los años 2005 y 2010 [50].

16

Figura 2.6: Niveles de precipitación y temperatura en Buenos Aires.

Si bien este no es un caso donde se utilice una gran cantidad de datos deentrada, podría repetirse este experimento calculando el promedio y varianzade los niveles de temperatura y precipitación registrados a lo largo de losaños, para cada día, y exponer los resultados en un gráfico similar.

2.3. MapReduce

El modelo de programación MapReduce, publicado por Jeffrey Dean ySanjay Ghemawat [2], surge como solución para el desafío de lidiar con lacomplejidad de sistemas de múltiples computadoras en concurrencia al pro-cesar cantidades masivas de datos. Se trata de una abstracción que permiteexpresar la lógica de las operaciones que trabajan sobre los datos, delegandoen una plataforma subyacente los aspectos de concurrencia tales como pa-ralelización del procesamiento, tolerancia de fallos, ubicación de los datos ybalanceo de carga.

Las solicitudes de procesamiento de datos que presentan los usuarios alsistema se denominan jobs. Desde el punto de vista de la abstracción pro-porcionada al usuario, la ejecución de un job procesa el archivo de entradaque ya se encuentra almacenado en un file system distribuido, dividido en

17

particiones dispersas por el cluster, para producir el resultado final.

Desde el punto de vista del sistema, cada job consiste de un conjunto detareas que pueden ser de dos tipos: las tareas Map, o MapTasks, se encargande ejecutar la función map() sobre los datos de entrada, y las tareas Reduce,o ReduceTasks, operan a través de la función reduce() sobre los productosintermedios para generar el resultado final. Ambas funciones son especifica-das por el usuario a la hora de ejecutar el job3. Las tareas son asignadaspor el Scheduler (componente de planificación) a los nodos del cluster paraoperar sobre los datos.

El concepto fundamental de este modelo es el de localidad de datos. Enla medida de lo posible, se intenta procesar los datos localmente, asignandolas tareas en los nodos donde están almacenadas las particiones del archivode entrada, evitando así transportarlas. De esta forma se minimiza el tiempode finalización de las tareas (dado que acceder a datos locales es más rápidoque transmitirlos por la red), al mismo tiempo que se mejora la carga detráfico en el cluster [52]. Otro aspecto importante que este modelo tiene encuenta es la tolerancia a fallas de los equipos. Al ejecutarse sobre clustersformados por múltiples nodos, es frecuente que existan fallas en algunos deellos.

Por lo tanto, MapReduce resulta ser un modelo que simplifica el desa-rrollo de procesos de explotación de datos al brindar una interfaz simple alusuario, mientras administra de forma automática las tareas, lidiando con lacomplejidad de la computación distribuida, para lograr optimizar el uso delos recursos disponibles y proporcionar tolerancia a fallos.

En lo que sigue de la sección se definirá desde el punto de vista formalel modelo MapReduce. Se presentará una descripción de la plataforma sub-yacente, encargada de ejecutar el procesamiento de los jobs sobre los datosde entrada a partir de las funciones Map y Reduce especificadas, y final-mente se realizará una comparación del modelo respecto a otros sistemas dealmacenamiento y procesamiento de datos de uso generalizado.

3Resulta interesante observar que este paradigma está inspirado en lenguajes funciona-les como Lisp, donde gran parte de las tareas de análisis de datos estriban en la aplicaciónsucesiva de dos primitivas, una llamada Map [51, p. 122] sobre los datos de entrada, yluego otra denominada Reduce [51, p. 145], sobre los resultados intermedios.

18

2.3.1. Definición del modelo

La ejecución de un job en MapReduce está dividida en dos fases, ambasejecutadas por múltiples procesos denominados tareas.

Durante la primera, llamada fase Map, se toma el archivo de entrada,previamente almacenado en el file system de base y dividido enM particionesque se encuentran distribuidas entre los nodos del cluster. Por cada partición,se ejecuta una instancia del método map() especificado.

Cada una de estas tareas, llamadas MapTasks, procesa la partición delarchivo de entrada que le ha sido asignada por el sistema. La interpreta,de acuerdo a la lógica definida por el usuario, como un conjunto de duplasclave-valor (k1, v1). Tras operar sobre estas duplas aplicando dicha función,se obtiene otro conjunto de duplas clave-valor de salida, (k2, v2).

Map : (k1, v1) −→ (k2, v2)

Las tareas Map son independientes entre sí, con lo cual la fase Map escompletamente paralelizable. Por otro lado, el desarrollo de la fase Reduceconsiste en la ejecución de R instancias del método Reduce, entre las cualesse reparten de forma equitativa las claves intermedias k2. Estas instancias sedenominan ReduceTasks.

Cada ReduceTask recoge todas las duplas (k2, v2), obtenidas de la faseanterior, cuyo valor de clave coincida con alguno de los que le fueron asig-nadas. Luego toma todas las duplas correspondientes a una cierta clave k2 ylas procesa mediante la función reduce(), obteniendo un valor final ν paraesa clave.

Finalizada la ejecución de todas las tareas Reduce, se obtiene como re-sultado de todo el job de MapReduce un valor final por cada clave.

Reduce : (k2, lista(v2)) −→ (k2, ν)

Dicho flujo de datos queda ilustrado en la figura 2.7. Es interesante notarque, si bien es posible la superposición de la fase Map con la recolecciónde productos intermedios por parte de las tareas Reduce, estas sólo puedencomenzar a procesar las duplas intermedias una vez que la fase Map finalizópor completo [2, 53, 54, 55].

19

Figura 2.7: Flujo de datos para un job de MapRecue [56].

Es decir que para aplicarse la función reduce() sobre los valores interme-dios correspondientes a una dada clave k2, es necesario que hayan finalizadode generarse todos los pares clave-valor intermedios producidos las tareasMap del job, y que la tarea Reduce hubiera terminado de recolectar todaslas duplas que le corresponden a esa clave (de forma de almacenar localmenteel total de su input).

2.3.2. Plataforma subyacente

El sistema sobre el cual se ejecutan los jobs de MapReduce es el encargadode lidiar con los aspectos de paralelización, tolerancia de fallos, ubicación delos datos y balanceo de carga, logrando así que resulten transparentes alusuario.

Al ser procesados por dicho sistema, los jobs codificados en estilo funcio-nal (especificando las funciónes Map y Reduce) son paralelizados de formaautomática y ejecutados en grandes clusters de computadoras de uso general.

La figura 2.8 muestra las dependencias entre los procesos que intervienenen la ejecución de un job de MapReduce. Los nodos que ejecutan las tareasMap leen la porción (split) de la entrada que les corresponde, y generan susresultados intermedios. Las tareas Reduce copian esos resultados intermediospara procesarlos y generar el resultado final del proceso.

20

Figura 2.8: Esquema de ejecución de jobs en MapReduce [2].

La gestión de los archivos es realizada por un filesystem distribuido. Tantolos input-splits que sirven de entrada para las tareas Map como las salidasgeneradas por las tareas Reduce, son almacenados en el (y tomados del)filesystem distribuido.

En el caso de Google MapReduce, el filesystem que funciona como soportesobre el cual opera la plataforma que ejecuta los jobs es GFS [57] (GoogleFile System).

De esta manera es posible enfocar un gran esfuerzo de ingeniería en opti-mizar la gestión de recursos por parte de este sistema-plataforma subyacente,una sóla vez, para luego aprovecharlo cada vez que se ejecuta un job sobrela misma.

2.3.3. Comparación con otros modelos y sistemas

En base a las características mencionadas del modelo MapReduce, es posi-ble notar que resulta especialmente útil como herramienta de procesamientobatch, dado que cada ejecución procesa el archivo de entrada completo paraobtener el resultado [58].

En contraposición a esto, las herramientas de análisis online, denomina-das OLTP, están orientadas a procesar un segmento particular de los datos

21

disponibles (generalmente, los más nuevos) con el objetivo de proveer resul-tados en tiempo real.

La diferencia entre MapReduce y otras herramientas o técnicas para elprocesamiento batch de datos es el avance que logra MapReduce en cuantoal tiempo para procesar cantidades masivas de datos, lo que permite ejecu-tar consultas ad-hoc sobre el set de datos completo, y experimentar con lainformación en tiempos razonables.

Comparado con los sistemas de bases de datos relacionales [59], MapRe-duce tiene la ventaja de poder lidiar con tamaños mayores de datos, yasean estructurados como desestructurados, en tanto que es posible escalarlode forma aproximadamente lineal, simplemente agregando nuevos nodos alcluster donde está implementado. Sin embargo, el modelo MapReduce estápensado para aplicaciones donde la información se escribe una vez y se leemuchas veces (ya que está orientado a optimizar el tiempo de lectura, másque el de escritura).

Por el contrario, los sistemas de bases de datos relacionales respondenen mejor tiempo cuando la cantidad de datos es menor y su naturaleza esestructurada, y además resulta de mayor utilidad cuando se necesita leer yescribir repetidas veces. Es por esto que los sistemas de bases de datos rela-cionales son un excelente complemento para MapReduce en las arquitecturasBig Data.

Resulta interesante señalar además que, si bien MapReduce es un mode-lo adecuado para una gran cantidad de problemas que requieren computardatos a gran escala, su esquema de procesamiento, que divide al job en dosfases, Map y Reduce, hace que para cierto tipo de aplicaciones se obten-ga performance sub-óptima y problemas de usabilidad, como en el caso deprocesamiento de grafos.

En este sentido el modelo de programación Pregel [60] surge para brindarun sistema distribuido de propósito general a gran escala, similar a MapRe-duce en el hecho de proporcionar un buen nivel de escalabilidad y toleranciafallas, pero orientado para el desarrollo de algoritmos de procesamiento degrafos, donde intervienen secuencias de iteraciones (que en Pregel se deno-minan supersteps).

22

2.4. Hadoop

Apache Hadoop es un entorno de procesamiento distribuido que surge co-mo implementación open-sorce de MapReduce. Su diseño está basado en unsistema que combina dicho modelo de procesamiento con GFS [57], esquemade filesystem distribuido publicado por Google.

Tiene su origen en Nutch [61], otro proyecto de Apache Software Founda-tion dedicado a la creación de un motor de búsqueda Web de código abierto,que forma parte a su vez de la librería de búsqueda de textos del proyectoApache Lucene. Tras la publicación de GFS y MapReduce, el equipo que par-ticipaba del desarrollo de Nutch encontró la posibilidad de implementar ensu proyecto las mejoras propuestas por el nuevo modelo de procesamiento,que les permitirían solucionar los problemas de almacenamiento y proce-samiento surgidos al buscar mejorar la performance (desideratum para laimplementación de un motor de búsqueda Web).

Dado que el sistema implementado en Apache para procesar informaciónbajo el modelo MapReduce podía ser aplicado en una gran cantidad de si-tuaciones que iban más allá de Nutch, se decidió separarlo como un proyectoaparte [58].

La capacidad de Hadoop para procesar datos a gran escala de forma efi-ciente, su calidad open-source y su diseño que hace transparentes al usuariolos problemas de bajo nivel, brindando al mismo tiempo una interfaz simple,empujó a una gran cantidad de organizaciones a utilizarlo buscando perfec-cionar sus procesos (algunos casos concretos pueden ser encontrados en labibliografía [58, p. 545] y en el sitio Web de Apache4). Esto generó a su vezuna importante comunidad de investigación y desarrollo que propuso mejo-ras y permitió perfeccionar el sistema, así como la aparición de un grupo deproyectos relacionados, conocidos en conjunto como Hadoop Ecosystem.

Entren los elementos de este ecosistema se encuentra Pig, entorno dedesarrollo para codificar procesos utilizando Pig Latin [62], lenguaje de altonivel para expresar operaciones sobre un flujo de datos, que luego se traducena MapReduce para ser ejecutados sobre Hadoop. Hive [63] es otro proyectodel ecosistema que implementa en Hadoop un data warehouse sobre el cualse pueden efectuar consultas escritas en SQL, a las que traduce para serprocesadas por la plataforma. Asimismo, HBase busca utilizar Hadoop paraobtener una base de datos distribuida. También es interesante mecionar el

4http://wiki.apache.org/hadoop/PoweredBy

23

caso de Mahout, herramienta montada sobre Hadoop para ejecutar procesosde Data Analysis.

Hadoop cuenta con una abstracción de la noción de filesystem, representa-da mediante la clase abstracta de Java org.apache.hadoop.fs.FileSystem[58, p. 52]. HDFS, la implementación de dicha clase que contempla un sis-tema distribuido de manejo de archivos, se desarrolla en profundidad en laSubsección 2.4.1, pero existen otras opciones. WebHDFS, por ejemplo, esuna implementación del sistema de archivos distribuido orientada a permitiraccesos vía Web para leer y escribir sobre una plataforma similar a HDFS.Por otro lado, KFS, también denominado CloudStore, es otra versión desa-rrollada en C++. Amazon desarrolló también su propia versión, S3, basadaen su servicio que lleva el mismo nombre.

Para proporcionar la funcionalidad del modelo MapReduce, Hadoop utili-za su implementación de filesystem en combinación con un runtime encarga-do de recibir las solicitudes de jobs que ingresan los usuarios, y procesarlasutilizando los recursos proporcionados por el sistema de base. La Subsec-ción 2.4.2 cubre la funcionalidad de dicho runtime.

2.4.1. Hadoop Distributed File System

Hadoop Distributed File System [64], abreviado HDFS, es una implemen-tación de la clase abstracta org.apache.hadoop.fs.FileSystem diseñadapara proveer a Hadoop de un sistema distribuido de gestión de archivos so-bre un cluster o conjunto de computadoras de bajo costo (de comercializaciónmasiva).

Se trata de un filesystem adecuado para aplicaciones que deben lidiarcon archivos de gran tamaño, en tanto que internamente HDFS divide losarchivos en particiones lógicas de longitud fija, cuyo tamaño es típicamente64 MB, que se denominan splits o bloques. Dado que los bloques de un mismoarchivo pueden almacenarse en distintos nodos, se tiene entonces que el límitede tamaño de archivo en este filesystem es la sumatoria de la capacidad dealmacenamiento de todos los nodos del cluster. Un archivo, entonces, serepresenta como una secuencia de bloques dispersos por el cluster.

La arquitectura de HDFS responde al modelo de comunicación denomi-nado master/slave. Uno de los nodos del cluster, al que se llama NameNode,cumple el rol de organizador del sistema de archivos ya que es el encargadode gestionar el espacio de direcciones [65, p. 173] del filesystem. Es además el

24

responsable del mapeo entre archivos y sus bloques, y de coordinar el accesoque hacen los clientes a los datos. El resto de los nodos cumplen el rol deDataNodes, en tanto que almacenan los bloques de archivos y ejecutan sobreellos las solicitudes que envían los clientes del filesystem.

De esta forma el NameNode almacena y manipula los metadatos relativosal uso del filesystem, pero la comunicación de datos se realiza directamenteentre el usuario y los DataNodes, por lo cual se evita un cuello de botella.Este tipo de arquitectura posibilita además un buen nivel de escalabilidad,ya que agregando nuevos nodos al cluster, el NameNode puede disponer desu espacio de almacenamiento y poder de procesamiento.

Una de las principales metas de diseño que se reflejan en la implementa-ción de HDFS es la de tolerancia a eventuales fallas. Mientras más grandesea el cluster de computadoras sobre el cual se está ejecutando Hadoop, ma-yor será la probabilidad de que ocurra una avería en alguno de los nodos.De hecho, en los entornos donde se utilizan computadoras y servidores decomercialización masiva, esta probabilidad no es nada despreciable.

Para sortear este problema, entonces, HDFS cuenta con un sistema dereplicación en el cual cada bloque puede ser almacenado en más de un nodo.El factor de replicación ρ se especifica a nivel de archivo, y determina lacantidad de réplicas que se crearán para cada uno de los bloques del archivo.Para el caso por defecto del sistema, donde se tiene ρ = 3, se suelen almacenardos réplicas en nodos separados de un rack, y la tercera en algún nodo enotro rack: de esta manera se evita la pérdida del bloque en el caso de queocurra una falla a nivel de rack en alguno de los dos.

La figura 2.9 muestra el funcionamiento del sistema de replicación enHDFS. En 2.9(a) puede verse que internamente el archivo es consideradouna secuencia de bloques (de igual tamaño), y cada uno de esos bloqueses copiado en múltiples nodos para garantizar su persistencia, además demejorar su disponibilidad. Asimismo, en 2.9(b) figura cómo el archivo, alser procesado o leído por un cliente, se reconstruye utilizando una réplicadisponbile de cada bloque.

Mediante un sistema de señales (llamadas “heartbeats”, o palpitaciones)enviadas periódicamente desde los DataNodes hacia el NameNode, es posibledetectar cuando suceden fallas en los nodos, racks o en la conectividad entreestos y el nodo coordinador. Ante la ausencia de dichas señales, el NameNodeasume que el nodo en cuestión está deshabilitado y por lo tanto toma losrecaudos necesarios para mantener el nivel de réplica necesario de todos

25

(a) Esquema de almacenamiento de réplicas.

(b) Reconstrucción del archivo tomando una réplica de cada bloque.

Figura 2.9: Sistema de replicación de bloques en HDFS.

los bloques almacenados allí, copiando sus réplicas hermanas en ubicacionesnuevas.

HDFS está diseñado además bajo el supuesto de que el sistema más efi-ciente de procesamiento es mediante el modelo por el cual los archivos seescriben una sóla vez y se leen muchas veces. Esto va en línea con los casosde uso más frecuentes en los problemas de Data Analytics, donde los datosson copiados inicialmente en el sistema para luego procesarlos sucesivas vecesrealizando distintos análisis. Al tratarse de archivos de gran tamaño, la tasade acceso a registros resulta más importante que la latencia al leer el primerregistro [58].

Es interesante observar además que en las versiones más nuevas de Ha-doop se introdujo el concepto de HDFS Federation para mitigar los límitesde memoria (y el cuello de botella impuesto a la administración de metada-tos) que acarrea el hecho de contar con un solo NameNode. Esto se lograseparando este rol en más de un nodo, cada uno de los cuales administrauna fracción del espacio de direcciones. Así se esquiva también el problemade tener un único nodo que controla el filesystem, robusteciendo el conceptode manejo de fallas en el cluster.

26

2.4.2. MapReduce runtime

El framework de MapReduce para Hadoop tiene por objetivo procesar lassolicitudes de jobs del usuario, para lo cual se vale de los recursos y funciona-lidades del filesystem de base, ya sea HDFS, WebHDFS, S3 o cualquier otraimplementación. En lo que sigue del capítulo, se analizará exclusivamente laversión clásica5 de este framework.

Al recibir la solicitud de procesamiento para un nuevo job de MapReduce,el sistema pone en marcha un mecanismo de ejecución (cuya especificaciónpuede ser hallada en la publicación de White [58], capítulo 6) que activaun componente denominado JobTracker, encargado de coordinar y gestionarla ejecución de los jobs. Este componente incorpora la solicitud a una listainterna que funciona como cola de espera, y en la medida que el sistema sedesocupe, inicializa el job agregándolo a otra lista, que funciona como colade servicio. La inicialización implica crear un MapTask por cada split delarchivo de entrada, y una cantidad de ReduceTasks igual a la pasada porparámetro por el usuario. Las tareas creadas son ejecutadas por otro tipode componentes, llamados TaskTrackers, que se ejecutan en los nodos delcluster.

En Hadoop, cada nodo cuenta con un número configurable de slots o cupospara tareas Map y Reduce. Estos cupos limitan la cantidad de tareas decada tipo que pueden ejecutarse simultaneamente en cada nodo del cluster.Cuando una tarea comienza a ejecutarse, ocupa un slot. Mientras esta tareaestá en ejecución, el slot se mantiene ocupado, hasta que al finalizar la tarea,se libera. De esta forma, cada slot puede ser asignado a lo sumo a una tareaa la vez.

La toma de decisiones acerca de qué tarea es asignada a cada slot delcluster (denominada scheduling) es realizada por el JobTracker, y será abor-dada detalladamente en la Subsección 2.4.3. Los TaskTrackers aprovechanel mecanismo de heartbeats para enviar periódicamente información al Job-Tracker sobre el estado de la ejecución. De esta forma, el JobTracker tieneacceso a información necesaria (como el progreso de cada tarea, y los no-dos que se encuentran disponibles esperando ser asignados) para gestionarel trabajo en el cluster.

5Existen ramas nuevas del proyecto, como la versión 0.23.0, en donde se incorpora unnuevo sistema, más general, denominado YARN, para el cual el runtime de MapReducees sólo un tipo de aplicación. En esta nueva versión, se da en llamar MapReduce 2 alruntime, y se menciona en la Subsección 2.4.4.

27

Figura 2.10: Flujo de datos para un job en Hadoop [58, p. 30].

La figura 2.10 ilustra el flujo de información en la plataforma Hadoop trasla ejecución de un job. Cada MapTask genera sus resultados intermedios ylos almacena en el disco local del nodo donde fue ejecutada. Al ser asignadaa un slot por el JobTracker, cada tarea Map procesa uno por uno de losregistros de su split asociado ejecutando el método map() definido por elusuario.

En el caso de que la tarea fuera asignada en una máquina que no guardaninguna réplica del split asociado, entonces el TaskTracker debe crear unacopia del split para luego recién ejecutar la tarea. En cambio, si se ejecuta enun nodo donde se encuentra almacenado el split que le corresponde, se diceque la tarea se ejecuta localmente: de esta forma se evita ocupar ancho debanda en el cluster y se ahorra el tiempo de copia. Naturalmente, Hadoopintenta en lo posible asignar las tareas donde serán ejecutadas de forma local(dicha lógica se encuentra encapsulada en el Scheduler).

Dado que las claves intermedias son repartidas entre las ReduceTasks deforma balanceada, cada una de ellas debe copiar todos los pares intermediosque correspondan a las claves que les fueron asignadas, en el disco local delos nodos donde se ejecutan, tomándolos desde las máquinas que ejecutaronlas tareas Map. En el caso general, cada tarea Reduce puede necesitar copiarresultados intermedios de cualquiera de las MapTasks (posiblemente de todasellas). En el ejemplo de la figura 2.10 puede verse un job con tres tareasMap y dos tareas Reduce, donde ambas tareas Reduce copian informaciónproducida por cada uno de los Mappers.

Para facilitar este proceso, las MapTasks particionan los resultados inter-

28

medios que generan, creando un segmento para cada Reducer. Este proceso,conocido como sort, es realizado por el TaskTracker que ejecuta la tareaMap, en simultáneo a su ejecución. Por otro lado, durante la etapa deno-minada shuffle, cada tarea Reduce copia en su nodo las particiones que lecorresponden, generadas por los Mappers. Una vez que la ReduceTask reco-lectó todas las particiones, que contienen productos intermedios para todaslas claves que le corresponden, debe entonces ejecutar un proceso de mergepara obtener un segmento por cada clave (dado que la función reduce()debe aplicarse separadamente, por clave, para todos los valores de cada unade ellas). Esta lógica queda esquematizada en la figura 2.11.

2.4.3. Scheduler

En Hadoop, el componente de scheduling cumple la función de asignarrecursos del cluster (slots desocupados) para las tareas Map y Reduce pen-dientes de aquellos jobs que se encuentran en la cola de servicio. Esta coor-dinación del trabajo en el cluster, de naturaleza distribuida, es por completotransparente al usuario de la plataforma.

No obstante, encierra un alto nivel de complejidad en tanto que hace nece-sario alcanzar un equilibrio entre requerimientos que a menudo se presentancomo contradictorios. (A saber, la optimización del tiempo de respuesta,del uso de la red y del poder de procesamiento, garantizar una distribuciónequitativa de recursos del sistema entre los distintos usuarios, entre otros.)

Sea J1, J2, · · · , Jq el conjunto de jobs a ejecutar, y sean el tiempo dedisparo ti,d y de finalización ti,f de cada job Ji, existen dos situaciones descheduling que hacen diverger su análisis. En el caso offline (o batch), setiene que los jobs son disparados conjuntamente en un tiempo inicial t0 (esdecir, ti,d = t0 ∀i), conociendo en aquel momento todas las propiedadesde cada job. Por el contrario, en el caso online se tiene que cada job Ji esdisparado de forma imprevisible en un momento ti,d, y su información no esconocida hasta entonces, por lo cual es más complejo diseñar algoritmos conbuen desempeño en estas circunstancias [66].

Formular métricas que describan los aspectos a optimizar de la performan-ce del Scheduler resulta necesario a la hora de comparar distintos algoritmos.Algunas métricas ampliamente utilizadas en la bibliografía son el makespano tiempo transcurrido entre la llegada del primer job hasta la finalizacióndel último job [67, 68], y el flowtime o sumatoria del tiempo transcurrido

29

Figura 2.11: Detalle de las etapas Shuffle y Sort [58, p. 178].

entre la llegada de cada job hasta la finalización del mismo [69] (obsérveseque minimizar el flowtime es equivalente a minimizar el tiempo promedio definalización de los jobs). También es útil tener en cuenta la localidad de datospara las tareas Map y Reduce (es decir, respectivamente, la proporción de ta-reas Map que se ejecutan localmente, y la cantidad de resultados intermediosque se deben transportar hacia los nodos que ejecutan los ReduceTasks).

Para intercambiar fácilmente distintas implementaciones del Scheduler enHadoop de acuerdo a las necesidades específicas de los usuarios, el frameworkcuenta con una clase abstracta TaskScheduler. Todas las implementacionesque se desarrollen en Java deben extender de esta clase abstracta, y parautilizarla en el sistema debe especificarse mediante la propiedad de configu-ración mapreduce.jobtracker.taskscheduler. A continuación se muestraun fragmento del archivo de configuración a modo de ejemplo.

Script 2.1: Fragmento del archivo de configuración de Hadoop

<property><name>mapred.jobtracker.taskScheduler</name><value>org.apache.hadoop.mapred.JobQueueTaskScheduler</value><description>The class responsible for scheduling the tasks.

</description></property>

El Scheduler seteado en este caso dentro del tag value es el default deHadoop. Es denominado FIFO (First In, First Out) y será abordado conmayor detalle en la Sección 3.1. Su implementación puede encontrarse enla clase JobQueueTaskScheduler del paquete org.apache.hadoop.mapred,

30

aunque el código fuente que se distribuye con la versión 1.2.1 de Hadooptambién incluye otras implementaciones como Fair Scheduler y CapacityScheduler.

Tal como fue mencionado en las secciones previas, la ejecución de un jobde MapReduce es un proceso en el cual se suceden distintas etapas. Durantela fase Map, cada MapTask ejecuta la función map() sobre el bloque quele corresponde (en caso de ejecutarse en un nodo que no posee una réplicade ese bloque, debe copiarlo al disco local), por lo que esta fase puede serconsiderada como una etapa. Por el contrario durante la fase Reduce, cadaReduceTask debe primero copiar todos los segmentos que le correspondande los resultados intermedios de las tareas Map, lo cual representa la faseshuffle/sort, y luego, una vez completada la recolección de resultados inter-medios, ejecutar sobre ellos la función reduce(), en lo que da en llamarse laetapa Reduce.

Cabe observar que durante las etapas Map y Reduce es intensivo el usode CPU para ejecutar las funciones predefinidas por el usuario sobre losbloques de entrada y los resultados intermedios, mientras que durante laetapa shuffle/sort el uso de la red (para transportar los resultados intermediosentre los nodos que ejecutaron las MapTasks que los producen, y los nodosdonde se ejecutan las ReduceTask que los procesan) representa el recursocrítico.

Si bien el comienzo de la etapa Reduce requiere necesariamente que fi-nalice la etapa Map, esta restricción no existe para la etapa shuffle/sort.De hecho, a medida que las distintas MapTasks producen sus resultados in-termedios, las ReduceTasks pueden comenzar a copiar los segmentos que lecorresponden.

Existe un mecanismo denominado early shuffle, por medio del cual secomienza a asignar las tareas Reduce de un job una vez completadas uncierto porcentaje (en general, 5%) de las tareas Map. Esto posibilita la su-perposición de las etapas Map y shuffle/sort, y tiene por objetivo encontrarun balance entre la carga de trabajo de la CPU y el uso de la red, aunquetambién hace más complicado el estudio del proceso [66].

El algoritmo de scheduling se encarga entonces de ubicar, en los slots delcluster que se encuentren disponibles, MapTasks y ReduceTasks pendientesde los jobs que se encuentran en ejecución, de forma tal de optimizar eltiempo de finalización de del conjunto de jobs y de cada job en particular, almismo tiempo que evita incurrir en problemas de inanición de ningún job.

31

En cuanto a la asignación de tareas Map en slots libres, deben tenerse encuenta aspectos distintos aspectos. Al mejorar la proporción de nodos queejecutan localmente sus tareas Map, se disminuye el tiempo necesario parafinalizar dicha tarea, mientras que se alivia al cluster de carga de tráfico porla reducción de comunicación de datos entre los nodos. Pero asignar todas lasMapTasks a máquinas que las puedan ejecutar localmente podría provocaruna distribución desequilibrada de trabajo entre nodos, desaprovechandoen cierta medida las capacidades del cluster en tanto que algunos estaríancargados de trabajo en exceso, mientras otros estarían osciosos [70].

Es por ello que la optimización del tiempo de resolución de un job requierealcanzar el mejor equilibrio entre localidad de datos y balance de carga enlos nodos. Asimismo, dado que cada MapTask ejecuta un bloque de datosde entrada (de tamaño fijo, típicamente 64 MB) por lo cual su duraciónes poco variable, y puesto que son tareas independientes, entonces resultaconveniente iniciar su ejecución tan pronto como sea posible.

Por otro lado, para la asignación de ReduceTasks debe tenerse en cuentala cercanía a los nodos que ejecutaron las tareas Map de ese job (y que por lotanto almacenan los resultados intermedios, producidos por ellas), entendidaen este caso como localidad de datos para las tareas Reduce, y la carga queesto implica sobre el uso de la red, dado que ejecutar todas las tareas reduceen un momento dado podría degradar la performance respecto a ejecutarlasgradualmente. Este concepto se desarrollará con mayor profundidad en elsiguiente capítulo.

Cabe observar también que en la medida en que evolucionen las tec-nologías de transmisión de datos en el interior del cluster, el algoritmo descheduling dejaría de ser un factor crítico para el éxito del modelo. Esto sedebe al hecho de que si el costo de transmitir datos entre nodos es irrelevan-te en términos de tiempo requerido o carga de tráfico, entonces asignar lastareas de forma balanceada entre los nodos del cluster sin tener en cuentala localidad de datos pasaría a ser la mejor estrategia de administración derecursos [71].

Sin embargo, existen estudios que demuestran empíricamente que en laactualidad la diferencia entre el ancho de banda de los discos locales y losaccesos a la red continua siendo significativa [52], y que por lo tanto lalocalidad de datos (tanto en las tareas Map como en la etapa shuffle/sort delas tareas Reduce) continúa siendo un factor determinante en la performancede Hadoop [53].

32

2.4.4. YARN

Con el objetivo de aprovechar el potencial de los sistemas operativos debase como HDFS, más allá del uso que se les da con el modelo MapRedu-ce, las versiones más nuevas de Hadoop reconfiguran su arquitectura paradisponer de un nuevo componente, denominado YARN (por “Yet AnotherResource Negotiator”). A esta nueva configuración se hace referencia me-diante el término Hadoop 2.0, en contraposición a Hadoop 1.0 que sería laimplementación clásica abocada al modelo MapReduce.

Hadoop YARN [72] busca entonces centralizar la administración de recur-sos del cluster con el fin de permitir la ejecución de distintas aplicaciones. Deesta forma el runtime original de MapReduce, delega en el nuevo componenteestas tareas para dedicarse exclusivamente a la lógica del procesamiento dedatos. En la figura 2.12 se esquematiza este cambio a nivel de los componen-tes del sistema.

A su vez, la versión del runtime de MapReduce que se adapta para ejecu-tarse sobre la nueva plataforma de Hadoop 2.0 figura en la literatura comoMapReduce 2, con el fin de distinguirla de su versión original.

Entre los requerimientos de diseño del YARN se cuentan los ya contem-plados en Hadoop 1.0, a saber escalabilidad, localidad de datos, toleranciaante eventuales fallas en los nodos y utilización eficiente de los recursos físicosdel cluster. Además, con la nueva configuración se busca lograr un manejoflexible de los recursos, un sistema dinámico de confguración del sistema ysoporte para diversos modelos de procesamiento en paralelo [73].

Figura 2.12: Reconfiguración de la estructura en Hadoop 2.0 con YARN.

33

Para cubrir estos requerimientos, se divide la funcionalidad que solía cum-plir el JobTracker en dos componentes separados que lleven adelante sus fun-ciones principales, pero que le permitan al sistema contar con la flexibilidadrequerida.

En primer lugar, el ResourceManager (RM) se encarga de realizar el segui-miento del uso de recursos de hardware por parte de las distintas aplicacionesy gestionar su asignación entre ellas. Además, por cada aplicación se cuentacuenta con un ApplicationMaster (AM), que efectúa su seguimiento particu-lar y coordinar la lógica de ejecución a nivel de job, solicitando rescursos alRM en la medida que hagan falta [72].

El diagrama de la figura 2.13 muestra la arquitectura de Hadoop 2.0.El RM se ejecuta como un proceso en un nodo dedicado, que actúa comoautoridad central en el arbitrio de recursos entre las distintas aplicaciónes.Se introduce la noción de contenedores, es decir porciones lógicas del totalde recursos en cada nodo.

Para monitorear dichos recursos, en cada nodo se encuentra en ejecuciónotro tipo de proceso denominado NodeManager (NM). La comunicación entreel RM y los NM se basa en heartbeats, como sucede en Hadoop 1.0. Cada NMes responsable de monitorear los recursos disponibles en su nodo, reportarlas fallas y administrar los contenedores locales [72].

Los clientes solicitan la ejecución de sus aplicaciones (generalización delos jobs de MapReduce) ante el RM, que evalúa la disponibilidad de recursosde forma similar a la versión original de Hadoop. Una vez que se cuenta conlos recursos suficientes, comienza su ejecución.

El ResourceManager posee también un plug-in para especificar la políticade scheduling que debe utilizar [73]. De esta forma, este tipo de plug-inspuede estar basado en los Scheduler descriptos en la Sección 3.1, Como FairScheduler, Capacity Scheduler y Coupling Scheduler.

En conclusión, el nuevo esquema permite utilizar Hadoop con otros mo-delos de programación, además de MapReduce, como Pregel [60], Dryad [74]y Spark [75], entre otros.

34

Figura 2.13: Diagrama de la arquitectura de Hadoop YARN [72].

2.5. Spark

Apache Spark es una plataforma de computación distribuída desarrolladaen el AMPLab de UC Berkeley, por el equipo de Matei Zaharia et al. [76].Pensada como una alternativa a Hadoop, cuenta con un esquema generalistaque permite abarcar distintos modelos de procesamiento masivo en paralelo(incluído MapReduce), y mejorar significativamente la performance en cier-tos casos de uso donde es posible mantener almacenados en memoria RAMlos datos a ser procesados.

Esta plataforma puede administrar un cluster de forma independiente(modo standalone), como así también es posible utilizarla sobre el YARNde Hadoop 2.0 o sobre Mesos [77], otro proyecto de Apache para el manejode clusters. Su arquitectura está inspirada en las implementaciones de Ma-pReduce, aunque basa su interfaz en la de DryadLYNQ [78] con el fin depresentar una API sencilla para el desarrollo de aplicaciones de alto nivel, ycuenta con versiones para Scala, Java y Python.

Si bien las implementaciones de MapReduce demuestran ser eficientespara una gran variedad de aplicaciones, existen ciertas situaciones en lascuales su desempeño no es óptimo. Esto dio lugar al desarrollo de modelosespecializados como Pregel [60], Piccolo [79] o GraphLab [80].

En particular, casos de uso tales como la aplicación de algoritmos itera-tivos de machine learning o la ejecución de jobs de análisis interactivo noencuentran en Hadoop y sus variantes buen rendimiento.

35

Los algoritmos iterativos suelen modelizarse en Hadoop como jobs con-secutivos, cada uno de los cuales representa una iteración. En cada una deellas, el job debe volver a cargar los datos (almacenados en disco) que fueronproducidos en la iteración anterior, incurriendo en un costo significativo quedegrada la performance de la aplicación.

Asimismo, la exploración interactiva de datos mediante consultas ad-hocsobre conjuntos de datos de gran tamaño, pero que podrían llegar a entraren memoria (considerando la capacidad sumada de los nodos del cluster)

Históricamente, la optimización de Hadoop ha estado orientada en mayormedida a la performance de jobs que realizan el completo del trabajo enuna sóla pasada de procesamiento en batch sobre datos que se encuentranalmacenados en disco [75].

En contraposición a esto, Spark introduce una estructura de datos nue-va, denominada Resilient Distributed Datasets (RDDs), que representa unacolección de objetos repartidos entre los nodos de un cluster [81]. Los RDDssurgen como extensión al modelo de MapReduce, con el objetivo de mejo-rar la eficiencia en el intercambio de datos entre etapas o jobs consecutivosy aportar generalidad al modelo de procesamiento [82]. Esta estructura dedatos es desarrollada en la Subsección 2.5.1.

Si bien en la actualidad el ecosistema de aplicaciones construídas sobreSpark no posee el mismo nivel de madurez que el de Hadoop, aquel se en-cuentra en contínuo desarrollo y crecimiento.

Spark SQL [83, 84], también conocido como Shark, es una herramienta dealto nivel para el procesamiento de datos estructurados mediante consultasexpresadas en SQL, HiveQL y Scala, y está diseñado de forma tal que seatotalmente interoperable con Hive. Asimismo, Spark Streaming es una adap-tación del modelo de programación D-Streams [85] que ofrece una extensiónde la API básica de Spark para soportar streaming de datos masivos a granvelocidad con tolerancia a fallas.

Además, MLlib es la librería de Spark para aplicaciones de machine lear-ning escalables, y posee distintos algoritmos y herramientas de aprendizajeautomático de uso común. GraphX, otra de sus librerías, abarca algoritmosde procesamiento de grafos e incluye una versión optimizada de la API quepresenta Pregel.

36

2.5.1. Resilient Distributed Datasets

En su tesis doctoral, Matei Zaharia presenta el concepto de Resilient Dis-tributed Dataset como una abstracción de estructura de datos que extiendeel modelo MapReduce, y tiene por objetivos lograr un transporte eficientede datos entre etapas sucesivas de procesamiento y generalizar los diversosmodelos de computación distribuída [82].

A fin de garantizar tolerancia a fallas de forma eficiente, el diseño deRRDs se basa en transformaciones de granularidad gruesa (coarse-grained)manteniendo un registro de las transformaciones aplicadas sobre los datasets.Dicho registro se denomina linaje, y sólo lleva cuenta de las transformacionesaplicadas, no de los datos resultados.

Puesto que la mayoría de las aplicaciones de procesamiento en paralelocontempladas por Spark aplican la misma operación sobre múltiples ítemsde datos, dicho enfoque de granularidad gruesa permite re-computar las par-ticiones que pudieran fallar durante el procesamiento, sin incurrir en costosmayores de almacenamiento (como ocurre en Hadoop o Dryad, que mantie-nen un grafo acíclico dirigido para recomputar eficientemente los procesosque fallan dentro de una iteración o job, pero agregan costos significativos alreplicar en el file system distribuído los datos resultantes de cada iteración).

En términos formales, los RDD son colecciones de registros u objetos,particionadas y de sólo lectura, y pueden ser creados aplicando operaciones(denominadas transformaciones) sobre datos almacenados en el file systemdistribuído, o bien aplicando transformaciones sobre otros RDDs.

Asimismo, no es necesario materializar un RDD hasta que sus datos seanrequeridos como input en una computación que busca devolver un resultadoal programa ejecutado, o hasta que sus datos deban ser almacenados en elfile system por especificación del usuario.

La interfaz utilizada para representar un RDD expone cinco elementosde información [81]. En primer lugar, un conjunto de particiones que formanlos elementos atómicos del dataset. Además, se cuenta con un conjunto dedependencias respecto a RDDs padres, con un iterador para computar unadada partición del dataset a partir de los iteradores de sus padres, y conmetadatos acerca del esquema de partición utilizado. Finalmente se cuentatambién con un operador preferredLocations(p) que devuelve una lista denodos donde la partición p puede ser accedida de forma más eficiente teniendoen cuenta el nivel de localidad de datos.

37

Las dependencias definen al RDD a partir de sus padres. Las mismasse clasifican de acuerdo a la cantidad de particiones del RDD (particioneshijas) que son generados a partir de cada uno de los bloques del padre o delos padres (particiones padres).

Se denominan dependencias exiguas, o narrow dependencies, a aquellasdonde cada partición padre es utilizada para generar a lo sumo una particiónhija. Este tipo de dependencias, por lo tanto, hace posible ejecuciones enforma de pipeline en el cluster. Además, ante el evento del fallo de unoo más nodos, para volver a calcular las particiones hijas perdidas sólo esnecesario obtener o re-computar las particiones padres que les corresponden.

Por el contrario, en las dependencias amplias, o wide dependencies, lasparticiones hijas pueden depender de más de una partición padre. Por estemotivo, para comenzar a computar las particiones del RDD hijo es necesarioque el cálculo de todas las particiones padres esté completo (imposibilitandouna ejecución en forma de pipeline).

Asimismo, recuperar particiones hijas tras la falla de algún nodo es menoseficiente en grafos cuyo linaje cuenta con dependencias amplias, dado quepara volver a calcular las particiones perdidas del RDD hijo, haría falta unare-ejecución completa.

En la figura 2.14 se clasifican algunas de las operaciones sportadas porSpark de acuerdo al tipo de dependencias que conducen, y se esquematizanejemplos de relación entre particiones entre RDD padres e hijos.

Figura 2.14: Tipos de dependencias entre RDDs [82].

38

Cabe destacar también que, dado el diseño de esta abstracción, resultanmás adecuadas para ser implementadas con RDDs las aplicaciones que efec-túan la misma operación sobre muchos registros de datos. Por el contrario,aplicaciones que actualizan, de forma asincrónica y con granularidad másfina, un conjunto de datos que representa un estado compartido, son menosadecuadas para este modelo.

2.5.2. Operaciones soportadas

Para garantizar la generalidad buscada en el diseño de Spark, el sistemacuenta con una serie de operaciones para ser aplicadas sobre los RDDs, queutilizadas de forma compuesta cubren la funcionalidad de la mayoría de lossistemas de computación distribuída especializados. Las operacines definidasse dividen en dos categorías: se denomina transformaciones a aquellas que seaplican sobre un RDD para generar otro nuevo (las cuales no necesariamentese computan de forma inmediata), y se denominan acciones a las operacionesque tienen por objetivo devolver un valor resultado al programa o almacenarlos datos en el file system.

Entre las transformaciones soportadas se encuentran por ejemplo flatMapy reduceByKey, que son equivalentes a las dos funciones fundamentales deMapReduce. Otras transformaciones son filter, que devuelve un subconjuntode los registros del conjunto de input de acuerdo a una función-filtro, sampleque devuelve un subconjunto aleatoreo del conjunto incial cuyo tamaño espasado como parámetro en forma de fracción y union que genera un RDDnuevo como la unión de dos existentes.

Además, groupByKey es una transformación que se aplica sobre RDDscuyos valores son de la forma clave-valor y devuelve un nuevo RDD dondetodos los valores que corresponden a una clave son agrupados (es decir quese trata un sub-proceso de la transformación reduceByKey).

Por otro lado, Spark cuenta con implementaciones de acciones tales comocount, que devuelve la cantidad de registros en un RDD, collect que devuelvetodos los elementos del RDD, lookup que busca un elemento en particularentre todos los del dataset de entrada, y save que guarda el RDD en unsistema de almacenamiento estable como puede ser el distributed file system.Además de dichas operaciones, el usuario puede controlar en qué momentodebe persistirse un RDD que se encuentra en memoria.

39

2.5.3. Persistencia en memoria de acceso aleatorio

En cuanto a la persistencia de RDDs, Spark ofrece tres posibilidades: al-macenamiento en memoria RAM (memoria de acceso aleatorio) como objetosJava sin serializar, almacenamiento en memoria RAM como datos serializa-dos, y almacenamiento en disco [81].

La opción de almacenamiento en memoria como objetos Java ofrece lamejor performance en términos de velocidad de lectura, dado que la máquinavirtual de Java puede realizar de forma nativa los accesos. En el caso dealmacenamiento en memoria como datos serializados, si bien las lecturasson menos veloces, se trata de un almacenamiento más eficiente en cuantoal uso de memoria (recurso habitualmente escaso). Finalmente, la opciónde almacenamiento en disco resulta necesaria para aquellos casos donde eltamaño de los RDDs es mayor al de la memoria, y demasiado costosos entérminos de procesamiento para ser re-computados durante cada uso.

La política de desalojo utilizada para administrar el espacio de memoriaes LRU (Least Recently Used). Al computar una nueva partición de un RDDen uso, en caso de no haber suficiente espacio en memoria, se desaloja algunapartición del RDD que más tiempo estuvo sin utilizarse excepto si se trata delmismo RDD al que pertenece la nueva partición [82]. De ser así, se mantienenlas particiones más antiguas del mismo RDD para evitar un comportamientocíclico e ineficiente de entradas y salidas de particiones.

Existe también la posibilidad de crear checkpoints de los RDDs en disco.Si bien mediante el linaje de un RDD siempre es posible recuperar cualquierade sus particiones en caso de ocurrir fallas en los nodos, este proceso pue-de resultar más costoso que la lectura de disco de todo el RDD en ciertoscasos donde el linaje se compone de cadenas largas o contiene numerosasdependencias amplias.

Para ello existe un flag, REPLICATE, que permite al usuario decidir quéelementos persistir mediante un checkpoint. Naturalmente, imponer check-points de forma frecuente garantiza una recuperación rápida en caso de fallas,pero degrada la velocidad de la ejecución. El equipo de desarrollo de Sparkbusca asimismo elaborar un sistema automático de checkpoints, utilizandoinformación disponible como el tamaño de un dataset y el tiempo que llevócomputarlo [81].

40

2.5.4. Scheduling y comunicación de datos

El scheduler de Spark construye un grafo acíclico dirigido (DAG, por sussiglas en inglés) de stages, o etapas, a ejecutar basándose en el linaje deun RDD [82], cuando el usuario ejecuta sobre el mismo alguna acción comocount, collect u otras descritas en la Subsección 2.5.2.

Se trata de un scheduler similar al de Dryad. Sin embargo, en Sparkse tiene en cuenta adicionalmente qué particiones de RDDs persistentes seencuentran almacenadas en memoria en los distintos nodos del cluster, conel objetivo de optimizar la performance explotando en lo posible la localidadde los datos y acortando los tiempos de acceso [81].

El sistema elabora dicho DAG de ejecución dividiéndolo en stages, deforma tal que cada stage contenga la mayor cantidad posible de transfor-maciones con dependencias exiguas. De esta forma, el scheduler lanza laejecución en paralelo de tareas para computar las particiones restantes decada etapa, hasta que el RDD objetivo esté completo.

Las dependencias amplias, sin embargo, constituyen un límite en la divi-sión de stages en el DAG debido a las operaciones de shuffle que requieren. Enla figura 2.15 se muestra un ejemplo de DAG de ejecución dividido en stages.Los bloques rellenos en azul y negro representan particiones, encontrándoseestos últimos en memoria. Los bloques de líneas sólidas que contienen par-ticiones representan los RDDs, designados por letras en mayúscula, y laslíneas discontínuas representan las etapas o stages.

Como puede verse en la figura, el sistema agrupa siempre la mayor can-tidad posible de transformaciones con dependencias exiguas en un mismostage, con el fin de aplicar ejecución en pipeline. Dado que en el caso de lafigura, el resultado de la etapa 1 se encuentra almacenado en memoria RAM(designado por el dataset B), se ejecuta la etapa 2 explotando la posibilidadde ejecución en pipeline. Finalmente se realiza la ejecución de la etapa 3 paraobtener el resultado.

Si en algún momento de la ejecución ocurriera una falla en alguna delas tareas, entonces se vuelve a ejecutar en otro nodo siempre y cuando lasetapas padres se encuentren aún disponibles. Si, por el contrario, el sistema seencuentra ante la falta de alguna de las etapas padres, entonces se vuelven alanzar en paralelo las tareas necesarias para volver a computar los elementosfaltantes.

41

Figura 2.15: DAG de ejecución de un job de Spark, divido en stages [82].

Durante las operaciones de shuffle (aquellas con dependencias amplias),Spark almacena los datos intermedios de las tareas en los nodos que las eje-cutan, de forma similar al tratamiento de resultados intermedios producidospor las MapTasks en MapReduce. Es posible de esta forma simplificar elproceso de recuperación de particiones ante la ocurrencia de fallas en losnodos.

En aquellos jobs de Spark donde se simula el comportamiento del modeloMapReduce, ejecutándose primero flatMap y luego alguna de las transfor-maciones similares a Reduce (reduceByKey, groupByKey o cogroup), se tienetambién una situación de shuffle. Para estos casos, en Spark debe especi-ficarse la cantidad de tareas Map y tareas Reduce que serán utilizadas, aligual que en Hadoop, aunque suelen utilizarse valores más grandes que en laconfiguración de esta otra herramienta [86].

Una diferencia significativa entre Spark y Hadoop, cuando ejecutan jobsque encuadran en el modelo MapReduce, es que la fase Reduce de Sparkrequiere que todos los bloques de la etapa shuffle entren en memoria al mo-mento que cada ReduceTask los necesita para ejecutarse. Cuando la memoriarequerida para dicha tarea es mayor a la disponible, se lanza una excepciónde memoria [86]. Para evitar estos problemas, se requiere especificar un va-lor elevado de cantidad de tareas Reduce, posiblemente a través sucesivasejecuciones de prueba y error.

42

Otras diferencias importantes entre ambas plataformas son el hecho deque en Spark actualmente no existe la posibilidad de solapar las etapas Mapy Reduce, a diferencia de Hadoop, y que las tareas Map en Spark construyen,cada una, un archivo por cada Reduce task objetivo, de modo que finalmenteuna ejecución conM tareas Map y R tareas Reduce deriva enM×R archivosintermedios [87].

A modo de optimización en este sentido, Aaron Davidson et al. [86] propo-nen utilizar un mecanismo denominado consolidación de archivos de shuffle,que compone los bloques de todas tareas Map ejecutadas en un mismo nododedicados a una cierta tarea Reduce. De esta forma, con valores típicos decantidad de tareas Map, tareas Reduce y nodos en el cluster, como podríanserM = 5,000, R = 1,024 y C = 16 respectivamente, en lugar de existir másde cinco millones de archivos intermedios, se obtienen 16,384.

2.5.5. Casos de uso de Spark y Hadoop/MapReduce

A continuación se detallan los casos de uso en los cuales la performance decada una de las plataformas es superior, a partir de los datos experimentalespublicados por diversas investigaciones sobre el tema. Se describen ademásalgunos ejemplos.

La performance de Spark es significativamente mejor en aquellos casosdonde la aplicación debe ejecutar sucesivas iteraciones sobre un mismo setde datos que cabe en el volumen total de memoria RAM del cluster. Este tipode algoritmos iterativos es utilizado frecuentemente en campos de aplicacióntales como procesamiento de grafos, optimización numérica y aprendizajeautomático.

En la figura 2.16 se exponen los resultados experimentales publicadospor Matei Zaharia et al. [76] para la ejecución de un algoritmo de regresiónlogística en ambas plataformas, utilizando un conjunto de datos de inputde tamaño menor al total de la memoria RAM del cluster. Las medicionesfueron registradas en condiciones idénticas de recursos asignados, y variandola cantidad de iteraciones que ejecutaba el algoritmo.

Es posible notar que la degradación de performance en el caso de Sparkes notablemente menor, dado que si bien la duración de la primera iteraciónes superior a la de Hadoop, las subsiguientes iteraciones tienen una duraciónsignificativamente más corta (dado que los datos utilizados entonces ya seencontraban en memoria).

43

1 5 10 20 300

1,000

2,000

3,000

4,000

Número de iteraciones

Tiempo

deejecuc

ión(s) Hadoop

Spark

Figura 2.16: Performance para un job con múltiples iteraciones en Spark y Hadoop(datos tomados de Zaharia et al. [76]).

En cambio, la duración de cada iteración en Hadoop es aproximadamentefija dado que los datos resultantes son almacenados en el sistema de archivosdistribuído, y necesitan ser leídos por la iteración siguiente.

Asimismo, Spark cuenta con un intérprete similar al de Python para reali-zar ejecuciones interactivas. Tras haber ejecutado una primer operación sobrealgún dataset (cuyo costo en tiempo es similar al de Hadoop), los bloquesdel mismo se encuentran ya disponibles en la memoria RAM de los nodos delcluster en caso de que el tamaño de los datos no sea mayor al del volumen dememoria total. Se tiene entonces que posteriores operaciones aplicadas sobreel mismo dataset podrán ser ejecutada con una duración considerablementemenor.

MapReduce, en cambio, obtiene mejor performance en aplicaciónes dondese ejecuta un job que realiza una sóla pasada de procesamiento sobre los datosde entrada [86], en especial si el tamaño total del input es mayor a la sumade recursos de memoria RAM de todos los nodos del cluster. Esto puede serobservado en la figura 2.16: para el caso de una sóla iteración, la performanceregistrada de Hadoop es mejor que la de Spark.

En la figura 2.17 se compara la performance de ambos sistemas para unjob con aquellas características, a partir de los resultados publicados porAaron Davidson et al. [86]. En el experimento, se leyeron datos almacenadosen S3, los mismos fueron parseados para formar pares clave-valor y luegoagrupados mediante la operación cogroup: a partir de un conjunto de pares

44

Spark Hadoop0

10

20

30

40

Tiempo

deejecuc

ión(m

in)

MapShuffleReduce

Figura 2.17: Performance para un job con una única iteración en Spark y Hadoop (datostomados de Davidson et al. [86]).

clave-valor (K,V ) con dos clases posibles V1 y V2 de valores, se agrupantodos los valores de una misma clave, separándolos en dos listas, una porcada tipo de valor. Finalmente, se almacenaron los resultado en HDFS.

Los siguientes capítulos de la presente tesis se abocan al estudio del mo-delo MapReduce implementado en Hadoop 1.0, sistema dedicado exclusiva-mente a dicho modelo. No obstante, el análisis y la optimización de la etapashuffle de MapReduce es también de interés para otras implementaciónes ypara distintos esquemas de flujo de operaciones distribuídas. En particularpara Hadoop 2.0, ya que cuenta con una implementación de MapReduce simi-lar a la de Hadoop 1.0 pero ejecutable sobre el YARN, y para Apache Spark,dado que sus operaciones distribuídas se ajustan al modelo MapReduce [76].

45

3

Estado del arte

Dada la popularidad alcanzada por el modelo MapReduce y la adopcióngeneralizada de Hadoop como herramienta para enfrentar los problemas deBig Data, en los últimos años se publicó una gran cantidad de trabajosdonde se estudian posibilidades de mejoras en la plataforma, para optimizardiversos aspectos del procesamiento masivo de datos: el rendimiento globalde la herramienta al ejecutar un conjunto de jobs, el rendimiento individualpor job, el tráfico en el cluster, el balance de carga de trabajo entre losdistintos nodos, el consumo de energía del sistema, entre otros.

El foco de la presente tesis está puesto en mejorar el algoritmo de schedu-ling a la hora de asignar ReduceTasks en los nodos del cluster para acortarlos tiempos de la fase shuffle/sort y finalmente acelerar la terminación de lastareas Reduce. De esta manera se optimizan conjuntamente el rendimientode Hadoop y el tráfico en el cluster.

A continuación se presenta el estado del arte de la investigación orienta-da al componente de scheduling de MapReduce/Hadoop en la Sección 3.1,incluyendo algunos conocidos algoritmos que se encuentran actualmente enuso. En la Sección 3.2 se describen los análisis de la optimización del uso derecursos relacionados a la asignación de tareas Reduce, hallados en la biblio-grafía de este trabajo. Finalmente, en la Sección 3.3 se mencionan trabajosque estudian aspectos tangenciales al abordado en esta tesis.

46

3.1. Implementaciones del Scheduler de Hadoop

Las decisiones tomadas por la plataforma respecto a la asignación detareas en los slots disponibles del cluster tienen un impacto determinantesobre su performance. Al ser un proyecto de código abierto, Hadoop permitediseñar algoritmos de scheduling enfocados en optimizar algún aspecto ométrica en particular de acuerdo a los requerimientos específicos que surjandel uso que se le planea dar a la plataforma.

El algoritmo de scheduling que trae Hadoop por defecto se denominaFIFO Scheduler. Se trata de un planificador sencillo, donde los jobs son pro-cesados por orden de llegada (sin tomar en cuenta prioridades ni el volumende trabajo de los mismos). Ante la llegada de un nuevo job, las tareas que locomponen son insertadas en una cola de la cual son tomadas para asignarlasa los distintos slots a medida de que estos se liberan.

Además del default FIFO Scheduler, existen otras implementaciones queserán descriptas a continuación. Esta sección describe la lógica de diseño delos algoritmos de scheduling a los cuales se tuvo acceso durante el desarrollodel trabajo.

3.1.1. Fair Scheduler

Debido al amplio beneficio que se obtiene en las plataformas de infor-mación al integrar en un sólo lugar el almacenamiento de los datos paraposteriores análisis, las organizaciones se enfrentan al problema de necesitarcompartirlos entre muchos usuarios. Dado que el componente de schedulingFIFO que trae por defecto Hadoop resulta poco adecuado para esta situa-ción, Matei Zaharia et al. [56] obtuvieron un diseño nuevo, llamado FairScheduler.

El objetivo de esta implementación es garantizar para los usuarios unamínima proporción de los recursos del cluster, y repartir aquellos que noestán siendo utilizados en un momento dado entre los usuarios que sí estánejecutando jobs, de forma proporcional a la fracción especificada para cadauno. De esta manera se aprovecha el total de los recursos al mismo tiempo quese dividen equitativamente. Los jobs se administran en grupos denominadospools. En principio, cada usuario tiene su propio pool, a donde van a parartodos los jobs que ejecuta. El balanceo de recursos se realiza a nivel de estosgrupos.

47

Las propiedades que se definen para cada pool son las siguientes: máximacantidad de jobs que pueden ejecutarse al mismo tiempo en ese pool, mínimacantidad de slots para tareas Map, y mínima cantidad de slots para tareasReduce. El algoritmo de reparto de capacidad se realiza, por separado, paraslots de tareas Map y tareas Reduce.

Si en un momento existe un solo job ejecutándose en el cluster, entoncesese job recibe el total de los recursos de procesamiento. En la medida quedistintos usuarios ejecuten otros jobs, entonces los nuevos slots libres sonasignados de forma tal de repartir el poder de procesamiento de maneraproporcional.

Así, los jobs pequeños pueden ser procesados en un tiempo razonable, in-dependientemente de que antes se hubieran ejecutado jobs de tamaño muchomayor. (Nótese que el principal defecto defecto del FIFO Scheduler para estetipo de aplicaciones, en donde se comparte el cluster entre muchos usuariosque pueden precisar realizar consultas pequeñas interactivas, es el tiempo derespuesta resultante para estas consultas pequeñas cuando son ejecutadasdespués de que se lanzaran jobs más grandes [56] que acaparan los recursosdel cluster por largo tiempo, dado que según el esquema FIFO no recibiríanatención hasta haber finalizado el anterior.)

En el ejemplo de la figura 3.1, donde la capacidad total del cluster es 100,ya sea en términos de slots para Mappers o Reducers, se tienen tres pools.Los pools 1 y 3 tienen respectivamente cuotas mínimas de 60 y 10 slots(puesto que el pool 2 no tiene establecido un mínimo, el algoritmo consideraque este es cero).

Figura 3.1: Ejemplo de alocación de recursos mediante Fair Scheduler [56].

48

Entonces, dado que el pool 3 no está ejecutando ningún job, sus slotspueden ser utilizados por los demás grupos. En tanto los jobs del pool 1requieran utilizar hasta 60 slots, estos le serán garantizados. Si necesitaranutilizar más de esa cantidad, entonces los 40 slots restantes que el clustertiene por asignar (y que no se están dedicando a satisfacer ninguna cuotamínima) serán repartidos de forma equitativa entre los pools que lo requieran.Para el caso ejemplificado en esta figura, los jobs del slot 1 requieren sólo 60slots, por lo cual el resto es asignado a los jobs del slot 2.

Adicionalmente, sus autores proponen un mecanismo para mejorar la lo-calidad de datos en la asignación de tareas Map, al que denominan DelayScheduling [56, 88]. Ante el arribo de un heartbeat con la señal de que unslot fue recientemente liberado, el mecanismo tradicional (en FIFO Schedu-ler) buscaba dentro de las tareas de este job alguna que fuera local en esenodo, y de no ser así buscaba entre esas mismas tareas alguna que tuvieralocalidad a nivel de rack. De no ser posible ninguno de estos dos casos, asig-naba cualquier tarea. El nuevo mecanismo propone buscar, si no hay ningunatarea local del job que se encuentra primero en la lista, entre las tareas de losjobs que le siguen, para priorizar la localidad de datos antes que el orden delos jobs. Pero pasado cierto tiempo en el cual un job fue salteado, entoncesasignar sus tareas de todas formas.

3.1.2. Capacity Scheduler

En el caso del Capacity Scheduler [89, 90], se cuenta con una serie de colasdefinidas por el administrador de la instancia, que funcionan de un modosimilar a los pools mencionados. Cada cola tiene asociada una capacidad,referente a los recursos (porcentaje de slots) de los que dispone, en tanto quelos jobs que pertenecen a una cola disfrutan como mínimo de esa cuota derecursos.

Cuando una de estas colas necesita exceder su capacidad, entonces elScheduler le asigna recursos de otras colas en la medida que estos esten libres.Este excedente puede ser reclamado de vuelta por la cola a la que pertenecíaoriginalmente, en el caso de que necesitase garantizar su capacidad.

Estas colas pueden ser jerárquicas, es decir, es posible replicar el sistemade reparto de recursos dentro de una misma cola. De no ser así, dentro decada cola los jobs se ejecutan en orden de llegada, como en el caso del FIFOScheduler.

49

Cabe observar que existe en Capacity Scheduler la opción de utilizar unsistema de prioridades a nivel de jobs montado sobre el esquema antedicho,en cuyo caso en cada cola se ejecutarán primero aquellos jobs con mayorprioridad, resolviendo los empates por orden de llegada.

Además, para prevenir que un usuario monopolice los recursos de la colaa la que pertenece, se establece un porcentaje límite sobre los recursos dela cola para cada usuario siempre y cuando existan más usuarios ejecutandosimultáneamente otros jobs el la misma cola.

Entre los distintos parámetros configurables para este scheduler, se des-tacan aquellos que refieren a los aspectos mencionados: la cuota de recursosdedicada para cada una de las colas, la implementación o no de prioridadesy el porcentaje límite dentro de una cola para sus usuarios.

Existen también parámetros relacionados al uso de memoria como porejemplo requerimientos mínimos de memoria para jobs específicos, es decirla posibilidad de ejecutar las tareas de aquellos jobs que tengan mayores re-querimientos de memoria, únicamente en nodos cuya capacidad de memoriaexceda cierto mínimo.

Comparado con Fair Scheduler, el diseño que propone Capacity Schedu-ler contempla un estricto control de acceso a cada cola [91]. Por ello, unainstancia de Hadoop utilizada por múltiples usuarios y que implemente Ca-pacity Scheduler puede ser vista como un cluster separado para cada grupode usuarios que comparten una cola, organizado internamente según el es-quema FIFO (y opcionalmente con un sistema de prioridades).

3.1.3. LATE Scheduler

El algoritmo de scheduling denominado Longest Approximate Time toEnd (LATE), publicado por Matei Zaharia et al. [54] tiene su foco en lare-ejecución especulativa de tareas: cuando Hadoop detecta que una tareaestá demorando más de lo normal, ejecuta de forma especulativa una copiade esta tarea en otro nodo, y tras terminada cualquiera de las dos, eliminala otra.

Mediante un análisis de los supuestos sobre los cuales se basan las deci-siones de estas re-ejecuciones especulativas en la versión original de Hadoop,y destacando que en ciertos casos (e.g. en servicios de cloud-computing) laheterogeneidad del hardware de los nodos que conforman el cluster provoca

50

que algunos de estos supuestos no se cumplan, LATE propone remediar estassituaciones volviendo a ejecutar especulativamente, siempre que encuentreslots disponibles, aquella tarea que estima finalizará última.

Para evitar que las tareas lanzadas por especulación se vuelvan a ejecutarsobre nodos poco performantes, el sistema tiene en cuenta la suma del pro-greso hecho en cada nodo para todas las tareas ejecutadas sobre ese nodo, yevita ejecutar estos re-lanzamientos sobre nodos que se encuentren entre losmás lentos (es decir, los que alcanzaron una menor sumatoria de progresohasta el momento).

La heurística propuesta por los autores para estimar el tiempo restantepara la finalización de una tarea dada [54] es simple y consiste en calcular latasa de progreso P como el cociente entre el nivel de progreso alcanzado porla tarea np (es decir, la fracción del total de trabajo de la tarea acabado almomento) y el tiempo transcurrido T , para luego obtener la estimación deltiempo restante tr a partir del cociente entre la fracción faltante de trabajode la tarea (i.e. 1− np) y la tasa de progreso.

P =npT

(3.1)

tr =(1− np)

P(3.2)

Del mismo modo, tomando en cuenta que volver a ejecutar tareas confines especulativos consume recursos del cluster, LATE incorpora un límitede tareas que pueden ser especuladas en un momento dado, y un sistema pa-ra determinar si una tarea es suficientemente lenta como para requerir unare-ejecución especulativa (mediante un nivel arbitrario denominado Slow-TaskThreshold).

Se tiene entonces que LATE es un algoritmo robusto ante la heterogenei-dad de la calidad de hardware de los nodos, en tanto que re-lanza especu-lativamente sólo las tareas más lentas (únicamente un pequeño número detareas), y prioriza aquellas que considera que degradarán en mayor medidala performance del job. Este algoritmo incluso evita ejecutar las tareas espe-culativas en nodos que detecta como poco efectivos. Cabe observar ademásque el diseño de LATE no es excluyente con el de los Schedulers presentadosanteriormente, por lo cual es posible ser incorporado a aquellos.

51

3.1.4. Resource-aware Adaptive Scheduler

Jordà Polo et al. [92] proponen un algoritmo de scheduling que decidedónde asignar las tareas basandose en un monitoreo general de los recur-sos del cluster, que denominan Resource-aware Adaptive Scheduler (RAS).Dichos autores consideran que el soporte de objetivos predefinidos por elusuario, y el monitoreo del nivel de utilización de recursos, son aspectos crí-ticos para Hadoop si bien fueron dejados de lado en otras implementacionesdel Scheduler.

Los objetivos de finalización de los jobs son establecidos al momento enque estos son lanzados. RAS maneja dichos objetivos como soft deadlines, esdecir metas flexibles, que sirven como guía para la toma de decisiones sobreel manejo de recursos.

Respecto al monitoreo del nivel de uso de recursos en el cluster, Polo etal. proponen modificar el modelo de slot (que en la versión tradicional deMapReduce están asociados únicamente a un tipo de tarea en particular, yasea Map o Reduce) por job-slots, que son específicos a un job determinadoy a un tipo de tarea en particular. El Scheduler, entonces, decide de formadinámica cuántos jobs-slots asignar en TaskTracker para un dado job ji deacuerdo a los objetivos predefinidos y requerimientos de recursos del job.

Estas decisiones se toman, además, teniendo en cuenta la capacidad derecursos del TaskTracker. Los tipos de recursos observados por la implemen-tación de RAS son el ancho de banda del disco duro, el tamaño de memoriay la capacidad de CPU del nodo que ejecuta el TaskTracker.

Entre otras aplicaciones interesantes de este enfoque, se tiene que la asig-nación de ReduceTasks puede tener en cuenta tanto las fases que se sucedendurante las tareas Reduce (es necesario notar que durante la etapa shufflese utiliza en gran medida la red, pero muy poco la CPU, y durante la etapareduce sucede el caso inverso) como el uso de aquellos recursos que se planeandedicar en el corto plazo para otras tareas a ejecutarse en el mismo nodo.

Por lo tanto, un control riguroso sobre el uso de los distintos tipos derecursos en los nodos podría aprovecharse en gran medida para optimizar elproceso de asignación de tareas Reduce (ya sea con el modelo de slots originalde Hadoop, o mediante el propuesto por los autores de RAS), y mejorar laperformance del sistema en tanto se logre un equilibrio entre procesamientoy comunicación entre los nodos.

52

3.1.5. COSHH

Otro esquema de scheduling es el presentado por Aysan Rasooli et al. [93],que dieron en denominar Classification and Optimization based Scheduler forHeterogeneous Hadoop Systems (COSHH). Este diseño busca considerar elproblema de heterogeneidad tanto a nivel del cluster, donde la calidad delhardware de los nodos puede variar considerablemente, como a nivel de losjobs que se ejecutan, que tienen distintos tamaños de trabajo y necesidadesde recursos.

En este trabajo se busca abordar simultáneamente las métricas de locali-dad de datos, reparto equitativo de recursos entre usuarios, cuota mínima derecursos asignados y promedio del tiempo de finalización. En el esquema dela figura 3.2 se muestra la arquitectura de alto nivel de COSHH, que se com-pone de cuatro modulos: (a) Hadoop System, (b) Task Scheduling Process,(c) Queuing Process, y (d) Routing Process.

El runtime de MapReduce presentado en la Subsección 2.4.2 conformael módulo Hadoop System de esta arquitectura. Por otro lado, el modulode COSHH llamado Task Scheduling Process tiene la función de estimar eltiempo promedio de finalización de los nuevos jobs que se ejecutan en elsistema, si se les dedicara cada uno de los recursos (nodos) del cluster. Estábasado en el desarrollo de Sameer Agarwal e Ion Stoica [94] denominadoChronos. Para que el esquema de COSHH resulte eficiente al ponerse enpráctica con cargas de trabajo reales, estos métodos de predicción debenresponder en cuestión de microsegundos y obtener resultados con un buennivel de certeza [93].

Figura 3.2: Relación entre componentes de COSHH [93].

53

Se tiene además que el módulo Queuing Process recibe la estimaciónobtenida en el paso anterior, y la utiliza en conjunto con otras característicasdel job entrante para clasificarlo. Cada clase de jobs se agrupa en una cola(del inglés, queue), y se actualiza el conjunto de colas agregando una nueva siel job nuevo no puede ser clasificado en ninguna clase. Asimismo, este módulose encarga de evaluar el perfil de cada clase de jobs, y dedicarle una porciónde los recursos según sus necesidades y según la disponibilidad del cluster.En la publicación de Rasooli et al. estas dos funciones se dan en llamarclasification-based approach y optimization-based approach, respectivamente.Ambas se ejecutan esporádicamente y a través de métodos veloces, con el finde no agregar una gran sobrecarga al proceso de scheduling.

Finalmente, en la medida que se liberen recursos (slots) que se encontra-ban prestando servicio, el módulo denominado Routing Process determinacuál de las clases de jobs utilizará el recurso recientemente liberado, entreaquellas sugeridas por la función de optimización del módulo antedicho. Lue-go de establecer la clase de jobs que utilizarán el recurso libre, se comunicaesta decisión al módulo Task Scheduling Process, que asigna este recurso aalguno de los jobs de dicha clase (véase el diagrama de la figura 3.2, quemuestra el flujo de información entre estos módulos).

3.1.6. Dynamic Priority Scheduler

Enfocados en la asignación automática de prioridades para el uso de re-cursos entre los distintos usuarios, según su determinación y capacidad deinversión, Thomas Sandholm et al. [95] publicaron un esquema de schedulingdenominado Dynamic Priority Scheduler (DP).

Según este esquema los usuarios tienen la capacidad de controlar la por-ción de los recursos que reciben, mediante el ajuste dinámico de la tasa degasto (spending rate), es el dinero que están dispuestos invertir por slot porunidad de tiempo. El algoritmo de scheduling, entonces, calcula en cada in-tervalo de tiempo la proporción de recursos que le corresponde a cada usuariocomparando su inversión con total de inversiones de todos los usuarios paraese intervalo.

Otro trabajo de los mismo autores [96] presenta distintas estrategias deasignación automática de prioridades y define métricas cuantificables paraevaluar su efectividad.

54

3.1.7. Coupling Scheduler

De publicación más reciente, Jian Tan et al. presentan un algoritmo descheduling que tiene en cuenta las dependencias entre tareas Map y Reducepara optimizar de forma conjunta su asignación, y al que dan en llamarCoupling Scheduler [97, 98].

Los autores observan que otros esquemas, en particular Fair Scheduler,logran optimizar la performance del sistema al mejorar la localidad de datospara las MapTasks, pero que por otro lado las ReduceTasks se asignan deforma separada (sin tener en cuenta las fuertes dependencias entre ambostipos de tarea) e ignorando para ellas el tema de localidad de datos, dejandolugar para implementar mejoras en este sentido.

Asimismo señalan que el mecanismo de Delay Scheduling [88], si bienmejora notablemente el nivel de localidad de datos para las tareas Map alelegir con mayor probabilidad nodos que las ejecuten localmente, introducedemoras que en ciertos casos derivan en una subutilización de los recursosdel cluster degradando la performance del sistema.

Buscando solucionar estas definciencias, Coupling Scheduler coordina laasignación de las tareas Reduce según el progreso de tareas Map, buscandouna optimización conjunta de su ubicación en los nodos del cluster. Para ellose incorporan dos mecanismos, a saber Wait Scheduling para la asignación deReduceTasks y Random Peeking Scheduling para la asignación de MapTasks.El diagrama de la figura 3.3 ilustra este diseño.

En lo que sigue se describen con mayor detalle dichos componentes y sereseña un modelo teórico (basado en la teoría de colas [99]) elaborado porlos autores para analizar las características del componente de scheduling enMapReduce y en particular la performance del scheduler propuesto.

Figura 3.3: Diagrama esquemático del Coupling Scheduler [97].

55

Wait Scheduling

La mayoría de las implementaciones del scheduler de MapReduce no con-templan la detención y posterior re-lanzamiento de tareas (en inglés, taskpreemption). Asimismo, existen implementaciones como Fair Scheduler quedisparan la ejecución de las ReduceTasks en tanto se completó un cierto por-centaje de la fase Map, de forma de superponerla con la etapa copy/shufflede la fase Reduce.

Sin embargo, ante la inexistencia del mecanismo de re-lanzamiento detareas, este enfoque puede incurrir en la inanición (starvation) de jobs en-trantes, en especial si se trata de jobs pequeños que arriban luego de otrosmás grandes [98]. Por lo tanto, la propuesta de Jian Tan et al. [97] de lanzargradualmente la ejecución de tares Reduce, de acuerdo al progreso de la faseMap, tiene por objetivo mitigar dichas situaciones de inanición.

Un efecto secundario de este método para asignar ReduceTasks de formaescalonada, que se menciona en el análisis de los autores, es el hecho de queno degrada la performance (en ciertos casos incluso la mejora) de MapReduceen situaciones en las que se procesa un único job en la plataforma.

Esto se debe a que un nodo ejecutando una tarea Map produce resulta-dos intermedios que se se almacenan en disco, y posteriormente las tareasReduce realizan un fetch de los segmentos que les corresponden. Por ello,los esquemas que disparan en un momento dado la ejecución de todas lasReduceTasks de un job aumentan la probabilidad de que estos se disputenel uso de recursos de los nodos que ejecutar las tareas Map y almacenanlos resultados intermedios. Como puede verse en la figura 3.4, este escena-rio denominado sync problem [98] puede solucionarse en tanto la ejecuciónescalonada de tareas Reduce disminuye la probabilidad de acceso simultáneo.

Finalmente cabe considerar que la asignación de tareas Map determinala ubicación de los resultados intermedios generados por ellas, mientras quela asignación de tareas Reduce (a mayor o menor distancia de una ciertacentralidad de todos los nodos que almacenan esos datos intermedios) afectaal costo de fetch en la etapa copy/shuffle. Es por esto que una progresióngradual en el lanzamiento de ReduceTasks, guiada por el progreso de la etapaMap, puede ayudar a optimizar el posicionamiento de las tareas Reduce dadoque aquellas asignadas más tarde tendrán mayor conocimiento de dónde seejecutaron nuevas tareas Map y por lo tanto acerca de dónde se encuentranalmacenados sus resultados intermedios.

56

Figura 3.4: Atenuación del problema de sincronización [98].

En conclusión, disparar de forma gradual la ejecución de tareas Reducedependiendo del progreso de las tareas Map mejora el rendimiento de la pla-taforma de acuerdo a los siguientes aspectos: alivia el problema de inaniciónaumentando la equidad de recursos para las tareas Reduce de los distintosjobs [100], permite regular la evolución de las fases Map y Reduce de formatal que la localidad de datos en la asignación de tareas de cada fase puede seroptimizada en forma conjunta, y disminuye el problema de sincronización enla etapa copy/shuffle al reducir la disputa de recursos.

Se tiene entonces que para un job i, se considera la fracción xi de tareasMap completas y la fracción yi de tareas Reduce en ejecución en un momentodado. Se asocia a cada job con una función fi (x) : [0, 1]→ [0, 1] de maneratal que ese job puede solicitar la ejecución de una nueva ReduceTask si ysólo si se cumple yi < fi (xi).

Para aquellos jobs que tienen tareas Reduce aún sin ejecutar, y cumplencon la antesmencionada condición, Coupling Scheduler computa una métricarelativa a la diferencia entre el progreso de las fases Map y Reduce, valor quees denominado mismatch. A mayor mismatch (es decir mayor desbalanceentre dichos progresos), un job tiene más derecho para solicitar la ejecuciónde tareas Reduce.

Aprovechando que las ReduceTasks de cada job son lanzadas de formagradual, aquellas disparadas más tarde tienen mayor conocimiento sobre ladistribución de almacenamiento en el cluster de los datos intermedios quedeberá copiar. Esta información es explotada por Coupling Scheduler paramejorar la localidad de datos de las tareas Reduce.

57

Algoritmo 1 Wait SchedulingInput: El nodo u ∈ V que declara un nuevo ReduceSlot libre.Output: Computa la asignación de tareas Reduce para el nodo u.1: if candidato = Null then2: (J,m0)← EvaluarCandidatos( )3: if m0 > 0 then4: candidato← J5: end if6: else7: wait← wait+ 18: i← bwait/Nc9: if i < 3 then

10: if (u ∈ Li(J)) y (u no está ejecutando ReduceTasks de J) then11: if se asigna con éxito una ReduceTask de J en u then12: candidato← Null13: wait← 014: else15: wait← i ·N + 116: end if17: end if18: else19: if se asigna con éxito una ReduceTask de J en u then20: candidato← Null21: wait← 022: end if23: end if24: end if

El Algoritmo 1 detalla la definición de Wait Scheduling tomada de lapublicación de Jian Tan et al. [98]. La función EvaluarCandidatos invocadaen la línea 2 calcula, entre los jobs que se encuentran en aquel instante queposeen al menos una tarea Reduce en la cola de ReduceTasks en espera paraser ejecutadas, el job J cuyo valor de mismatch m0 es el mayor, y retornaambos elementos.

Dadas la topología del cluster G = (V,E) compuesta por un conjunto Vde nodos y un conjunto E de conexiones entre ellos, una métrica de distanciah(u, v) entre dos nodos u, v ∈ V y la cantidad de datos intermedios wi(u) enu para un job i.

58

Sea además Vr ⊂ V el conjunto de nodos del cluster con ReduceSlotsdisponibles, entonces para el job i se define en la ecuación 3.3 el costo totalde transferencia de datos intermedios a un nodo u ∈ Vr.

Ci(u) =∑v∈V

wi(v)h(u, v) (3.3)

Para decidir dónde asignar un ReduceTask solicitado, Coupling Schedulercalcula Ci(u) para cada u ∈ Vr y retiene los K = max(Vr, D) de menor valor(es decir más centrales), siendo D una constante arbitraria, e.g. D = 7. Losnodos correspondientes asociados a estos valores retenidos son almacenados,en orden creciente, en una lista L(i).

Dicha lista L(i) puede ser accedida mediante tres interfaces distintas:L0(i) que contiene únicamente al primer elemento de la lista original, L1(i)que contiene al segundo y tercero, y finalmente L2(i) que contiene a losdemás. Estas son utilizadas en la línea 10 del Algoritmo 1. De esta formase define centralidad al nodo que figura primero en dicha lista, dado queminimiza Ci(u).

En cada ronda de comunicación entre el master node y los N slave nodes,donde se reciben los N heartbeats que llegan en orden aleatorio, CouplingScheduler genera la lista L(i) para cada job i con ReduceTasks solicitadas,y de priorizando entre jobs a partir a su mismatch, asigna sus tareas Reduceen el nodo más central de L(i).

La figura 3.5, tomada del trabajo de los autores, expone una comparaciónde la performance de este esquema respecto a Fair Scheduler e ilustra ellanzamiento paulatino (y ligado al progreso de la fase Map) de las tareasReduce.

Figura 3.5: Comparación entre Coupling Scheduler y Fair Scheduler.

59

Random Peeking Scheduling

Si bien la localidad de datos para las tareas Map determina en granmedida la performance de la plataforma, existe un compromiso respecto ala utilización de recursos: es necesario encontrar un balance entre forzar laejecución de MapTasks en nodos con localidad de datos, y evitar situacionesde subutilización de recursos [98]. Es decir, en ciertos casos donde los recursoscomputacionales abundan, lanzar tareas Map incluso en nodos donde no seejecutan de forma local puede ser beneficioso.

Random Peeking Scheduling toma como base cualquier algoritmo que,ignorando la cuestión de localidad de datos, decida qué MapTasks corres-ponde lanzar para garantizar la equidad (fairness) entre los jobs y usuarios,por ejemplo Fair Scheduler [56] o FLEX [101]. Dicho algoritmo proporcionauna lista Lm de tareas Map candidatas a ser ejecutadas.

Ante la llegada de un nuevo hearbeat, de encontrarse libre un MapSlot,se toma una tarea Map de la lista y se ejecuta en ese slot con probabilidadpj según la ecuación 3.4. Si el nodo podrá ejecutar la tarea de forma local(es decir, si contiene almacenados input splits del job en cuestión por serprocesados), entonces el valor que toma pj es 1. De lo contrario, si la tareano será ejecutada de forma local, el valor que toma pj es menor a uno.

pj =

1 ejecución con localidad

1− αj(pjm)βj (

1− e−Nm

)en otro caso

(3.4)

En este último caso, Random Peeking Scheduling ejecuta la tarea deforma aleatoria teniendo en cuenta los siguientes parámetros: la fracción pjmde nodos en el cluster que tienen almacenados input splits del job j, el númeroNm de nodos que tienen disponibles MapSlots y la cantidad M j

p de tareasMap pendientes para ese job.

De esta forma se calcula pj favoreciendo que el scheduler decida saltear elheartbeat en caso de que muchos otros nodos tengan input splits para ese job(y que podrían ejecutar las tareas map de forma local). Asimismo se favoreceque el scheduler opte por asignar la tarea, a pesar de la falta de localidad dedatos, si la cantidad M j

p de tareas Map pendientes de ese job aumenta encomparación con el número Nm de nodos con MapSlots disponibles.

60

Las cantidades αj y βj que figuran en el cálculo de pj se obtienen segúnlas ecuaciones 3.5 y 3.6 respectivamente, y reflejan la necesidad de posicionarlas tareas Map cerca de sus input splits, así como de las tareas Reduce en eje-cución. De esta manera, y en colaboración con el algoritmo Wait Scheduling,se optimiza de forma conjunta la ubicación de tareas Map y Reduce.

αj =

0, 7 localidad a nivel de rack y ReduceTask en nodo0, 8 localidad a nivel de rack1 en otro caso

(3.5)

βj = 0, 1 + 0, 9(

1− e−Mj/maxNm,1)

(3.6)

Nótese que para calcular αj , se asigna un valor menor (aumentando laprobabilidad pj de lanzamiento de la tarea Map) en caso de existir localidadde datos a nivel de rack para esa tarea, y en caso que el nodo que alberga elMapSlot libre esté ejecutando también una tarea Reduce.

En conclusión, el scheduler ignora el heartbeat de un nodo con algúnMapSlot recientemente liberado con probabilidad 1− pj , esperando obtenermejor localidad de datos en el siguiente heartbeat. Esto da lugar a un procesode Bernoulli, cuya esperanza de hearbeats sucedidos hasta que se asigna latarea Map es 1

pj[102].

Modelo teórico

Basándose en las investigaciones de Predrag R. Jelenković et al. sobrelas distribuciones de variación regular, que forman parte del grupo de distri-buciones heavy-tailed [103], y en particular sobre la disciplina de servicio decolas denominada Processor Sharing [104], los autores de Coupling Schedulerelaboran un modelo teórico a fin de analizar la performance del scheduler ycompararlo con el esquema de Fair Scheduler.

De acuerdo a la nomenclatura de Kendall [105] y en el contexto del modeloMapReduce, una cola M/GI/1 modeliza el arribo de jobs a un servicio (porejemplo, el procesamiento de la fase Map) de acuerdo a un proceso de Poissoncon tasa λ, y el tamaño de los jobs, medidos de acuerdo a la cantidad detareas que lo componen, son variables aleatorias, independientes entre síe identicamente distribuídas, iguales en distribución a B. De considerarseestable, la carga ρ , λP [B] de la cola debe satisfacer ρ < 1.

61

Al llegar a la cola, el job comienza a ser atendido inmediatamente. Deacuerdo a la disciplina de servicio Processor Sharing (PS), la tasa de servicioque recibe cada uno es inversamente proporcional a la cantidad de jobs enla cola. Es decir, en caso de haber n jobs, cada uno recibe una proporciónigual a 1

n de la capacidad total de servicio.

Se dice que la distribución de una variable aleatoria B es de variaciónregular con índice −α (notado B ∈ Rα) cuando cumple con la ecuación 3.7,donde l(x) : R+ → R+ es una función slowly varying, i.e. lımx→∞

l(ηx)l(x) = 1,

con η > 1.

P [B > x] =l(x)

xα, x ≥ 0, α ≥ 0 (3.7)

Jian Tan et al. [106, 107, 100] proponen entonces modelar el sistemade scheduling en MapReduce mediante los siguientes elementos. En primerlugar, el arribo de jobs es representado por un proceso de Poisson dondelos intervalos de tiempo Aii<−∞ entre la llegada de jobs consecutivos seasumen de distribución exponencial con tasa λ.

En segundo lugar, el servicio que ejecuta la fase Map de los jobs es repre-sentado por una cola M/GI/1/PS (processor-sharing queue [108]), siendo elconjunto Bii<−∞ de variables aleatoreas i.i.d. del tiempo requerido parafinalizar toda la carga de trabajo de tareas Map para cada job i en casode contar ese job con la exclusividad de los recursos del cluster. Estudiosempíricos sobre el comportamiento de clusters de producción de MapReducemuestran que la duración de los jobs y el tiempo entre arribos responden adistribuciones heavy-tailed [109]. Para el modelo en cuestión, se asume quela distribución de las variables Bi es de variación regular (es decir, cumplencon la ecuación 3.7).

Por último, en cuanto al modelado del servicio de la fase Reduce, se tienela cantidad r > 0 de ReduceSlots en el cluster y Ri el número de tareasReduce para cada job i. Se caracteriza mediante la tupla (Cji , D

ji ) la j-ésima

tarea Reduce del job i, siendo Cji la duración de su etapa shuffle/sort, y Dji

la de su etapa reduce.

Finalmente, cabe destacar que si bien Coupling Scheduler es una propues-ta inteligente e innovadora, continúa sin resolver completamente la situaciónde inanición donde jobs largos le quitan recursos a los jobs pequeños [110].

62

3.2. Optimizaciones en la asignación de Reduce-Tasks

A diferencia de las tareas Map, que trabajan sobre un split que se en-cuentra enteramente almacenado en una misma máquina, las tareas Reduceutilizan como input una porción del resultado intermedio producto de cadaMapTasks del job (de todas ellas en el caso más general) que se encuentraalmacenado donde la tarea Map fue ejecutada.

Uno de los principios rectores del modelo MapReduce, orientado a mejo-rar su performance, afirma que llevar el proceso computacional hacia dondese encuentran almacenados los datos resulta más economico que realizar eltransporte de los datos para copiarlos en los nodos que los procesarán.

No obstante, mientras que la localidad de datos para las tareas Map seencuentra ampliamente estudiada y optimizada en las implementaciones máspopulares del Scheduler, para las tareas Reduce este problema recibió menoratención a pesar de su potencial impacto [55].

Si bien sucede que una cantidad considerable de investigaciones sobre laoptimización de MapReduce pasa por alto el hecho de que los datos inter-medios necesitan ser transportados hacia los nodos que ejecutan las Redu-ceTasks [53], en los últimos años surgieron trabajos que buscan perfeccionarel proceso de decisión que implica asignar tareas Reduce.

Esta sección desarrolla las propuestas orientadas en este sentido. En laSubsección 3.2.1 se presentan los principales enfoques que sugieren el usode algoritmos greedy para asignar tareas Reduce en los slots que se encuen-tran más cerca de la mayoría de los resultados intermedios que ya fuerongenerados.

Un mecanismo estocástico de asignación de las ReduceTasks propuestopor Jian Tan et al. que busca minimizar el costo total de transporte de re-sultados intermedios es presentado en la Subsección 3.2.2, entretanto en laSubsección 3.2.3 reseña una heurística de decisión basada en dicho mecanis-mo, pero que simplifica notablemente su implementación.

Finalmente, la Subsección 3.2.4 detalla un algoritmo de scheduling pa-ra las tareas Reduce propuesto por Xiaotong Zhanga et al. basado en ladisciplina de asignación shortest remaining time first (SRT).

63

3.2.1. Enfoques greedy

A continuación se presentan los principales esquemas para la asignaciónde ReduceTasks orientados al uso de algoritmos greedy [111]. La estrategia deasignación utilizada en dichos esquemas consiste en recorrer secuencialmenteuna lista de tareas Reduce a ser asignadas, y al momento de evaluar cadauna de ellas, se las asigna al slot disponible que minimice el costo de fetchde los resultados intermedios.

Center-of-Gravity

El algoritmo Center-of-Gravity Reduce Scheduler (CoGRS) presentadopor Mohammad Hammoud et al. [112] busca extender, para las tareas Re-duce, el principio que sugiere acercar el procesamiento hacia los datos, queen las versiones iniciales de Hadoop se encuentra restringido para las tareasMap.

Resultados experimentales de dichos autores indican que en ciertas ejecu-ciones de MapReduce el tamaño total de los resultados intermedios es igualal del archivo de input (por ejemplo, en algunas implementaciones de algo-ritmos de ordenamiento) e incluso mayor (en algoritmos de clustering comoK-means).

Shadi Ibrahim et al. [113] observan además que existe una varianza signifi-cativa tanto en la distribución de claves intermedias entre las tareas Reduce,como en la distribución en los nodos del cluster de particiones correspon-dientes a una dada clave intermedia a lo largo del cluster.

En consecuencia, el diseño de Hammoud y su equipo busca ubicar cadatarea Reduce Ri en su centro de gravedad, determinado por la localizaciónen el cluster de las particiones que le corresponden a dicha tarea y ponderadode acuerdo al tamaño de cada una de estas particiones (cuya distribución,como fue mencionado, sufre de una significativa varianza).

Para ello introducen una métrica que denominan distancia total ponde-rada en la red (WTND, por sus siglas en inglés), calculada a partir de laecuación 3.8. Obsérvese que se utiliza también en Coupling Scheduler.

WTNDi(u) =∑v∈V

wi(v)h(u, v) (3.8)

64

Aquí se calcula la sumatoria, sobre cada uno de los v del cluster quealmacenan particiones dedicadas a la ReduceTask Ri —conjunto V—, delproducto entre el tamaño wi(v) de dicha partición y la distancia topológicah(u, v) que divide al nodo u que ejecutaría aquella tarea y al nodo v quealmacena la partición. De esta forma, como puede verse en la ecuación 3.9,se designa al centro de gravedad de Ri, notado COG(Ri), como el nodo queminimiza WTNDi(u).

COG(Ri) = mın∀uWTNDi(u) (3.9)

Para determinar con precisión el nodo COG(Ri) es necesario conocer lalocalización de todas las particiones de Ri. Esto es posible, únicamente, unavez finalizada la ejecución de todas las tareas Map del job.

Ahora bien, en los casos en donde la instancia cuenta con el mecanis-mo early shuffle activado, esta información no se encuentra disponible almomento de asignar las tareas Reduce. Por este motivo, en dichos casos elalgoritmo admite calcular COG(Ri) con los datos disponibles una vez fina-lizado cierto porcentaje α (no necesariamente 5% como el valor por defaultpara early shuffle en la versión original de Hadoop) de las MapTasks, y lan-zar la asignación de tareas Reduce una vez completo ese porcentaje de lafase Map.

El trabajo de los autores sugiere establecer, para una dada aplicación deMapReduce, dicho porcentaje α mediante la evaluación de la performancede una serie de ejecuciones de prueba o datos históricos. A este porcentajese refieren como sweet spot, o punto óptimo. Así, se sacrifica precisión en elreconocimiento delcentro de gravedad real de Ri, pero se logra aprovecharlas ventajas de early shuffle en cuanto a las mejoras en la performance delsistema.

El algoritmo intenta entonces de ubicar las ReduceTasks en su centro degravedad. Para ello, al recibir la señal de un TaskTracker ejecutándose en unnodo v que posea un ReduceSlot libre, verifica si alguna de las tareas reducepor ejecutarse posee ese nodo como centro de gravedad, y de ser así la asignaa ese nodo.

De encontrar más de una tarea Reduce cuyo centro de gravedad es v,entonces se prioriza aquella tarea cuya partición almacenada en ese nodo esla de mayor tamaño. En cambio, si no encuentra ningúna, asigna aquellatarea cuya distancia topológica h (COG(Ri), v) sea mínima.

65

LARTS

Para mejorar la performance de MapReduce mediante la asignación deReduceTasks, buscando minimizar la cantidad de datos transmitidos durantela etapa shuffle/sort, es necesario llevar un seguimienton de la ubicación y eltamaño de la partición de los resultados intermedios que le corresponden acada una de estas tareas. Locality-aware ReduceTask Scheduling (LARTS) esun mecanismo de asignación de tareas Reduce orientada hacia dicho objetivo,publicada también por Mohammad Hammoud et al. [114].

Dado que cada ReduceTask puede llegar a recibir una partición de cual-quier tarea Map del mismo job, la ubicación (y el tamaño) de todas lasparticiones de los productos intermedios que generan las MapTasks no seconoce con precisión hasta el momento en que termina de ejecutarse la fa-se Map. Luego de completada esta fase, se encuentra disponible de formacompleta la información sobre el paradero y las dimensiones de dichas par-ticiones, brindando mayor inteligencia a la hora de elegir dónde ejecutar lastareas Reduce.

No obstante, el Scheduler de Hadoop cuenta con un dispositivo opcionaldenomimado early shuffle, presentado en la Subsección 2.4.3. Early shufflecomienza a asignar las ReduceTasks de un job una vez alcanzado un ciertoporcentaje de tareas Map finalizadas, y con ello reduce el tiempo de ejecu-ción del job mediante el trabajo simultáneo de la fase Map y de la etapashuffle/sort de la fase Reduce.

Para optimizar la performance MapReduce, existe por lo tanto un com-promiso entre dos aspectos: asignar las tareas Reduce lo más temprano posi-ble, maximizando el solapamiento entre las fases Map y Reduce, y asignarlascontando con información completa sobre la ubicación y el tamaño de lasparticiones, maximizando las chances de reducir el trafico en la red durantela fase shuffle/sort.

El esquema de LARTS propone entonces, para aquellos jobs que son eje-cutados de forma periódica sobre archivos de entrada similares, realizar unanálisis de sensibilidad sobre ciertas variables que serán presentadas a con-tinuación, para optimizar la performance de su algoritmo.

Se define en primer lugar una variable ES que designa la fracción detareas Map de un job que deben finalizar su ejecución antes de comenzar aasignar las tareas Reduce. Esta variable se ajusta observando la performancedel algoritmo durante pasadas ejecuciones, buscando alcanzar un cierto nivel

66

óptimo, denominado sweet spot [114], que permita convinar las ventajas deearly shuffle con la inteligencia para asignar tareas Reduce lograda al contarcon suficiente cantidad de información sobre las particiones.

Una vez dispuesto a ejecutar una tarea Reduce Ri de un dado job i,LARTS utiliza la información disponible para reconocer cuál es el rack delcluster que maximiza el volúmen de datos que le corresponden a dicha tareade acuerdo a sus particiones, y cuál de entre los nodos de este rack es el quemaximiza la misma cantidad. Se utiliza MaxRack(Ri) y MaxNode(Ri) paranotar, respectivamente, dichas cantidades.

Los TaskTrackers informan periódicamente, mediante heartbeats, sobresu situación de disponibilidad tanto respecto de sus MapSlots como de susReduceSlots al JobTracker que administra el trabajo del cluster. Sea entoncesNode(TTj) el nodo que ejecuta el TaskTracker j, y Rack(TTj) el rack dondese encuentra dicho nodo.

LARTS considera que la mejor decisión que puede tomar sobre la asigna-ción de una ReduceTaskRi es asignarla en aquel nodo que cumplaMaxNode(Ri).Inicialmente, intenta asignar la tarea de esta manera, rechazando las solici-tudes de cualquier TaskTracker que no se encuentre en ejecución en ese nodo.

Dado que rechazar demasiadas solicitudes de los TaskTrackers podríaincurrir en una degradación de la performance del sistema, se aplica un con-tador C(TTj) que se incrementa cada vez que LARTS rechaza un heartbeatdel TaskTracker TTj solicitando que se le asigne una tarea Reduce. Una vezque este contador alcanza cierto umbral α, se permite un relajamiento dela condición antes mencionada, y se permite asignar en dicho TaskTrackercualquier ReduceTask tal que Rack(TTj) = MaxRack(Ri). En caso de queC(TTj) continúe creciendo, entonces a TTj se le asigna de forma aleatoriacualquier tarea Reduce que esté lista para ser ejecutada.

Para lograr la mejor parametrización del algoritmo, se registra la perfor-mance variando los valores de α y ES. Los autores señalan que, en la medidaen que se incrementa el valor de α, aumenta la demora en la asignación detareas Reduce y disminuye el nivel de utilización del sistema y paralelismo,pero al mismo tiempo se libera de tráfico el cluster.

Por otro lado, en tanto que se aumenta ES, disminuye el solapamientoentre la fase Map y la etapa shuffle/sort, pero se reduce también el traficoen el cluster, ya que el algoritmo posee mayor información y puede realizarmejores decisiones.

67

Purlieus

Diseñado para optimizar la performance de los sistemas MapReduce brin-dados como servicio en la nube, Purlieus [55] propone abordar el problemade localidad de datos desde el momento en que se almacenan inicialmentelos archivos de entrada en el cluster. Para ello, se busca ubicar las réplicas delos splits de archivos de entrada teniendo en cuenta sus consecuencias sobrela cantidad de datos transportados durante la etapa shuffle/sort.

Este enfoque, publicado en 2011 por Balaji Palanisamy et al., clasificaincialmente los archivos que ingresan al sistema para ser almacenados deacuerdo a los jobs que los utilizarán como input, para tomar distintas es-tratégias a la hora de elegir los nodos donde almacenar los splits iniciales ydonde asignar tareas Map y Reduce.

En todos los casos, el objetivo está puesto en minimizar el costo totalde translado de datos a traves del cluster. Para ejercer esta clasificación, losautores se valen de las diversas técnicas de profiling para jobs de MapReduce,que consideran efectivas en tanto resultan ser veloces y precisas [115, 116].

Las clases de jobs consideradas en Purlieus son: (a) Map-input heavy jobs,es decir aquellos en los que el temaño total de los resultados intermediosgenerados por las tareas Map es negligente respecto al tamaño del archivode entrada, (b) Map-and-Reduce-input heavy jobs, en los cuales el tamaño delarchivo input del job es similar a la sumatoria del tamaño de los resultadosintermedios, y (c) Reduce-input heavy jobs, aquellos donde el conjunto deresultados intermedios supera ampliamente en tamaño al archivo de entrada.

Modelando según la ecuacuón 3.10 el costo total Cost(i, d) de transportede datos para un job i ejecutado sobre un set de datos inicial d el equipode Palanisamy evalúa, para cada una de esas clases de jobs, cuál es el factorpredominante que debe ser atendido para minimizar el resultado de la suma.

Cost(i, d) = Mcost(i, d) +Rcost(i, d) (3.10)

En esta ecuación,Mcost(i, d) representa el costo de transporte durante lafase Map de dicho job, es decir, el costo de transporte de bloques del datasetde entrada d para ser procesados por las tareas Map que se ejecutan sinlocalidad de datos, y Rcost(i, d) representa el costo de fetch de resultadosintermedios por parte de todas las tareas Reduce del job, esto es, el costo detransporte durante la fase Reduce.

68

Se tiene entonces que, en aquellos jobs perfilados como (a) Map-inputheavy, es el factor Mcost(i, d) quien determina el costo total de transportepara el job. Es por ello que Purlieus desestima en estos casos la localidad dedatos para las tareas Reduce, y se centra en distribuír el trabajo de la faseMap entre los nodos del cluster de forma equitativa.

Para ello, ubica inicialmente los bloques de d entre los nodos en los quese espera menor carga de trabajo (teniendo en cuenta los bloques de otrosdatasets que ya almacena) y asigna, siempre que sea posible, las tareas Mapen nodos que las ejecutarán con localidad de datos. De no ser esto posible,busca nodos con MapSlots libres en orden creciente de cercanía a dondese encuentran almacenados los bloques, desempatando según la medida decarga de trabajo esperada.

Los jobs que se categorizan como (b) Map-and-Reduce-input heavy, porotro lado, poseen valores de Mcost(i, d) y Rcost(i, d) del mismo orden. Elesquema propuesto, entonces, busca optimizar ambos de forma simultánea. Atal efecto, los bloques de datos son ubicandos inicialmente en sub-sectores delcluster, suficientemente grandes como para albergar al job, y que maximicenla interconexión entre nodos (denominados n-clubs [117]).

Con ello se logra que, en la medida en que se consiga asignar la mayorparte de Map- y ReduceTasks dentro de aquel sub-sector del cluster, enton-ces los resultados intermedios generados por las tareas Map se encuentrenubicados a menor distancia de los nodos quee ejecutan las tareas Reduce.

Por último, los jobs que entran en la categoría (c) Reduce-input heavytienen un valor de Rcost(i, d) dominante en la ecuación 3.10, y en estos casoses más importante abordar la localidad de datos para las tareas Reduce. Laubicación de bloques del archivo de entrada se realiza en cualquier nodo delcluster con un buen nivel de espacio de almacenamiento disponible, en tantoque resulta despreciable el caudal de transporte de dichos bloques hacia losnodos que ejecutan las tareas Map.

Para reducir el valor de Rcost(i, d), maximizando la localidad de datosdurante la etapa shuffle/sort de la fase Reduce, Purlieus intenta asignar enestos casos la mayor cantidad de tareas Map y Reduce en un n-club especí-fico dentro del cluster. Si bien las tareas Map deberán entonces transportarlos bloques para procesarlos, según el precedente análisis este trafico no esimportante en comparación a la ganancia que se logra al disponer de todoslos resultados intermedios en nodos cercanos a los que ejecutan las tareasReduce del job.

69

Coupling Scheduler (asignación de ReduceTasks)

Como ya fue expresado en la Subsección 3.1.7, el mecanismo de asignaciónde tareas Reduce de Coupling Scheduler [97, 106] consta del lanzamientogradual de las ReduceTasks de un job dependiendo del progreso de las tareasMap, permitiendo ubicar a las primeras lo más cerca posible de su centralidad(aquel nodo que minimice Ci(u) según la ecuación 3.3).

Dicho lanzamiento escalonado de tareas Reduce puede observarse en losgráficos de la figura 3.6. De esta forma se perfecciona el rendimiento de laplataforma en tanto que permite:

Aliviar el problema de inanición en las situaciones donde el arribo dejobs grandes es seguido por el de jobs pequeños, aumentando la equidadde recursos para las tareas Reduce de los distintos jobs.

Regular la evolución de las fases Map y Reduce de forma tal que lalocalidad de datos en la asignación de tareas de cada fase puede seroptimizada en forma conjunta.

Reduce el problema de sincronización durante la etapa copy/shuffle alminimizar la disputa de recursos.

Se trata de un algoritmo greedy dado que, una vez propuesta una tareaReduce para ser ejecutada, se evalúa en ese momento cuál es el nodo queminimiza la función de costo Ci(u) y que tenga al menos un slot libre, asig-nando la tarea en dicho nodo. En este sentido, el concepto del algoritmo essimilar a Center-of-Gravity Reduce Scheduler.

Figura 3.6: Progreso de Map- y ReduceTasks en Coupling Scheduler [98].

70

3.2.2. Asignación secuencial estocástica

Jian Tan et al. [53] proponen un modelo de scheduling para optimizarla ubicación de tareas Reduce que se basa en la teoría clásica de asignaciónsecuencial estocástica.

Si bien los enfoques greedy representan un avance significativo respecto alas versiones iniciales de Hadoop (que ignoran la localidad de datos durantela etapa shuffle/sort), es posible obtener una mayor optimización del sistemasi se toma en cuenta la probabilidad de arribo futuro de tareas Reduce denuevos jobs en mejores condiciones para aprovechar los recursos a disposición.

Dada una serie S de n jobs que aparecen en forma secuencial, donde cadajob j tiene asociado una variable aleatoria Xj (siendo estas independientesentre sí e idénticamente distribuídas) que toma un valor xj .

Sea además un conjunto T de n recursos disponibles, cada uno de loscuales se corresponde con un valor p, que cumple 0 ≤ p ≤ 1, y numeradosde forma tal que p1 ≤ p2 ≤ · · · ≤ pn.

Considerando la necesidad de asignar un recurso para cada job en elmomento en que este aparece (cada uno de estos momentos consiste en unaetapa de asignación) y que aquellos recursos pueden ser utilizados sólo unavez, y dado que la recompensa que se espera obtener si se decide utilizar eli-ésimo recurso para cumplir con el job j es igual al producto pixj , se tieneque la esperanza del beneficio obtenido en el caso de utilizar un esquema deasignación ξ(·) : S → T está dada por la ecuación 3.11.

E

n∑j=1

pξ(j)Xj

(3.11)

Entonces el problema de asignación secuencial estocástica consiste en ma-ximizar el beneficio obtenido durante aquel ejercicio de decisión.

Para dicho problema, Cyrus Derman et al. [118] demuestran que la solu-ción óptima consiste en calcular a partir de la ecuación 3.12, para cada etapade asignación k, una sucesión de valores −∞ = a0,k ≤ a1,k ≤ · · · ≤ ak,k =∞,para luego asignar el i-ésimo recurso al job j si y sólo si xj se encuentra con-tenido en el intervalo (ai−1,k, ai,k].

71

ai,k+1 = ai−1,k FX(ai−1,k)

+ ai,k [1− FX(ai,k)] +

∫ a1,k

ai−1,k

z dFX(z)(3.12)

Estos valores ai,k pueden ser calculados de forma recursiva, y dependende la función de distribución acumulada de las variables aleatorias Xj , cuyadistribución es idéntica (i.e. FX(x) = P [Xj ≤ x] para cualquier j.).

Basándose en este resultado, el equipo de Jian Tan plantea una versióndel problema clásico de asignación estocástica adaptada al componente descheduling de MapReduce. En esta versión adaptada del problema, cada jobj ∈ S, cuya variable aleatoriaXj está relacionada a la cantidad total de datosintermedios producidos durante su fase Map, tiene asociado una cantidadRj (también aleatoria) de tareas reduce. Por lo tanto, el job j consume Rjrecursos en lugar de uno sólo.

Por otro lado, cada recurso i ∈ T representa un ReduceSlot, libre almomento de la llegada del job, y su valor pi está asociado a la distanciatopológica acumulada del recurso, es decir la suma de las distancias entre elnodo que alberga al recurso y todos los demás nodos del cluster. A los efectosde mantener como objetivo del problema el hecho de maximizar la ecuación3.11, puede tomarse como p al valor inversamente proporcional normalizadode la distancia topológica acumulada.

Sean además Bj la carga de trabajo del job j para su fase Map, r lacantidad total de ReduceSlots del sistema, y r la cantidad máxima de tareasReduce que puede tener un job, y sea K la cantidad máxima de jobs quepueden estar en ejecución simultáneamente en el cluster. Se tiene tambiénque Xi

j es cantidad datos intermedios que le corresponden a la i-ésima tareaReduce del job j, por lo cual se cumple,

Xj =

Rj∑i=1

Xij , ∀j.

Inicialmente, desarrollan un esquema de asignación óptimo asumiendoque la carga de trabajo de la fase Map de todos los jobs es muy grande oinfinita (e.g. Bj =∞). Por consiguiente, se tiene que el sistema procesa sólolos primeros K jobs que llegan. Si bien se trata de un escenario poco realista,sirve como base para el análisis del caso más general.

72

Cuando el j-ésimo job arriba al sistema (con 1 ≤ j ≤ K), se encuentracon un conjunto S(j) de l(j) slots libres. Ordenandolos según su distanciatopológica acumulada, de forma tal que δjk sea el k-ésimo slot libre de menordistancia acumulada que encuentra el job j al arribar al sistema, se tiene que

S(j) =δj1, δ

j2, · · · , δ

jl(j)

.

Si se cumple Kr ≤ r, entonces se asegura la disponibilidad de la canti-dad de tareas reduce necesarias para cada job, al momento que comienzansu procesamiento. Para cada job que arriba en j-ésimo lugar, que encuen-tra l(j) ReduceSlots libres al momento de arribo, el esquema de asignaciónóptimo consiste entonces en calcular, a partir de las ecuaciones 3.13 y 3.14,la sucesión de valores dada por ∞ = q0,j ≥ q1,j ≥ · · · ≥ ql(j+1)+1,j = −∞(estos valores umbrales, o thresholds, son independientes de las distancia to-pológicas acumuladas). De esta forma, se le asignan al job j los Rj slotslibres numerados δjy, δjy+1, · · · , δ

jy+Rj−1 de S(j) si y sólo si se cumple que

qy,j >Xj

Rj≥ qy+1,j .

Para el cálculo de los thresholds mediante las ecuaciones 3.13 y 3.14 debetomarse q0,n = ∞ y ql(j+1)+1,n = −∞ para n ∈ 1, 2, · · · ,K − 1, y con elvalor q1,K = −∞ para el caso particular del K-ésimo job, definiendo ademáslas funciones G(x) = P

[Xj

Rj≤ x

]y G(x) = P

[Xj

Rj> x

]para cualquier j.

• Si i > r,

qi,n−1 = qi,n G(qi,n)

+

r∑a=1

P [R = a]

(∫ qi−a,n

qi,n

z dG(z) + qi−a,n G(qi−a,n)

) (3.13)

• Si 1 ≤ i ≤ r,

qi,n−1 = qi,n G(qi,n) +

(r∑a=i

P [R = a]

)∫ ∞qi,n

z dG(z)

+

i−1∑a=1

P [R = a]

(∫ qi−a,n

qi,n

z dG(z) + qi−a,n G(qi−a,n)

) (3.14)

73

En los sistemas de producción reales, sin embargo, el supuesto por el cualla carga de trabajo de los jobs tiende a infinito es poco realista. Se tiene,por el contrario, que los jobs arriban aleatoriamente y parten tras finalizarsesu procesamiento. Para estudiar este nuevo escenario, el equipo de Jian Tanasume que las distribuciones Bj de carga de trabajo de cada job j es i.i.d.respecto de la de los demás jobs, y que los jobs arriban de acuerdo a unproceso de Poisson de tasa λ.

A continuación se introducen algunos elementos de notación para des-cribir el proceso de asignación óptimo para este caso más general. Dadoun conjunto A, sea P(A) el conjunto de partes de A, es decir el conjuntoformado por todos los subconjuntos de A, incluyendolo así como tambiénal conjunto vacío ∅. Sea además Pk(A) el mayor subconjunto de P(A) cu-yos elementos tienen todos exactamente tamaño k. Defínase, para cualquierconjunto A formado por nodos del cluster, la función H(A) como la suma-toria de la distancia topológica acumulada de todos los elementos de A, i.e.H(A) =

∑v∈AH(v).

Dado el conjunto conjunto S(j) de slots libres al momento del arribo deljob j al sistema, y para un valor arbitrario v ≥ 1, sea L(j) la lsta de todos loselementos de Pv(S(j)) ordenados de menor a mayor utilizando como métricala función H(·) cuando |L(j)| > 0.

Entonces, si λE [B] y Rj = v, el esquema de asignación óptimo en estecaso consiste en asignar al job j los Rj slots libres contenidos en el y-ésimoconjunto de la lista L(j) si y sólo si se cumple, para una cierta sucesión dethresholds ∞ = q0,j ≥ q1,j ≥ · · · ≥ q|L(j)|+1,j = −∞, la inecuación 3.15.

qy,j >Xj

Rj≥ qy+1,j (3.15)

Para este escenario más realista, con arribos y partidas de los jobs, losthresholds o valores umbral qi,k1≤k≤|L(j)|+1 dependen no sólo de las distri-buciones del tamaño de los datos intermedios Xj y de la cantidad de tareasreduce Rj , sino además de la tasa de arribos y de las distancias topológicasacumuladas. Si bien el trabajo no finaliza el desarrollo de la política de asig-nación óptima para este último escenario, dada la complejidad para calcularlos thresholds mencionados, se elabora una heurística basada en este análisis,que es presentada a continuación.

74

3.2.3. Heurística RHCP

Receding Horizon Control (RHC), también denominado Model PredictiveControl (MPC), es un método de control de procesos en el cual las decisionesde control se obtienen optimizando una dada métrica, de manera on-line encada instante de muestreo, mediante un modelo del sistema que, en virtudde obtener resultados en una cantidad de tiempo razonable, se limita a uncierto horizonte de predicción [119, 120].

En el trabajo citado en la Subsección 3.2.2 se presenta también una heu-rística denominada Receding Horizon Control Policy (RHCP) que se basa endicho método. Agregando ciertos supuestos restrictivos al análisis de asigna-ción de tareas Reduce expuesto en aquella subsección, Jian Tan et al. [53]logran simplificarlo notablemente para dar lugar a esta política de decisión.

La heurística RHCP resuelve el esquema de asignación óptimo para elcaso donde K = 2, es decir un sistema que evalúa optimizar la asignación derecursos únicamente para el job que acaba de arribar, y el siguiente, en basea información disponible sobre los útimos τ jobs arribados anteriormente alsistema.

Asumiendo que tanto la distribución del intervalo de tiempo Aj entre lallegada del job j y el siguiente, como su carga de trabajo Bj de la fase Map,son exponenciales con tasas λ y µ para cualquier job j considerado, se tieneentonces que la ecuación 3.16 determina la probabilidad p de que el primerjob siga en ejecución cuando el segundo job arriba al sistema [99], definiendoademás ϕ = λ/µ.

p =λ

(λ+ µ)=

ϕ

(1 + ϕ)(3.16)

Utilizando la misma notación que en el apartado anterior, defínase la listaΥ(j) de valores del conjunto S(j) de slots libres al momento de arribo deljob j ordenados de forma creciente a partir de la función H(·) de distanciatopológica acumulada para cada nodo del cluster.

Considerando además que la cantidad de ReduceTasks de ambos jobs esla misma, i.e. Rj = Rj+1 = R, se obtiene entonces una heurística simplebasada en un sólo threshold: asignar a j los primeros Rj slots de la listaΥ(j) si se cumple que,

75

Xj ≥ Rjϕ

(1 + ϕ)E[X

R

]. (3.17)

En caso contrario, asignar al recién llegado job j los slots de Υ(j) inde-xados según Rj +1, Rj +2, · · · , 2Rj . Luego, ante la llegada del siguiente job,j + 1, asignar a este los primeros Rj+1 valores de la lista, en cualquiera delos dos casos.

El cálculo del threshold involucrado en esta política de decisión requiereestimar los valores de ϕ y E [X/R]. Para ello se propone un esquema adapta-tivo que actualiza de forma dinámica el valor de estos parámetros, de acuerdoa la información disponible de los jobs útimos τ arribados.

Ante la llegada un nuevo job i, el sistema captura la cantidad Ni dejobs presentes en ese momento, el número Ri de ReduceTasks y el cáclu-lo de la cantidad de datos intermedios Xi que serán producidos por el job(lo cual puede observarse, por ejemplo, tras ejecutar una pequeña cantidadde MapTasks e interpolando linealmente la cantidad de resultados inter-medios obtenidos). Esta información es almacenada entonces en una listaW = (Ni, Ri, Xi)1≤i≤τ de tamaño τ que acumula los valores wi = (ni, ri, xi)para cada uno de los últimos jobs arribados al sistema.

Así es posible obtener valores estimados para el promedio de longitud dela cola N , y para el promedio de datos intermedios por tarea Reduce Y ,según las expresiones

N =1

τ

τ∑k=1

nk, Y =1

τ

τ∑k=1

xkrk. (3.18)

Asimismo, a partir de la teoría de colas se tiene que la esperanza de lalongitud de la cola puede ser calculada a partir de la expresión 3.19 [53].

E [N ] =ϕ

(1− ϕ)(3.19)

Entonces, tomando E [N ] = N efectuando el reemplazo de ϕ según 3.19,y sustituyendo Y por X/R en el cálculo del threshold involucrado en 3.17,se obtiene el método para computar la heurística RHCP ante la llegada decada job.

76

3.2.4. Algoritmo SRT

Xiaotong Zhang et al. [110] proponen una versión del método de schedu-ling Shortest Remaining Time (SRT) adaptado a la asignación de recursosen el modelo MapReduce: para elegir, entre los jobs en ejecución, de cualde ellos tomar las ReduceTasks a asignar, se comparan las estimaciones deltiempo restante de la fase Map de los mismos.

A diferencia de la solución de Coupling Scheduler, para lanzar gradual-mente las tareas Reduce de acuerdo al progreso de las tareas Map, el algo-ritmo SRT busca asignar las tareas Reduce de acuerdo al tiempo restantedel job. Por ello, para que el método sea eficiente necesita contar con esti-maciones precisas del tiempo que resta para completar todas las tareas Mapde cada job en ejecución.

Sean pj y fj las cantidades de tareas Map pendientes y finalizadas, res-pectivamente, para el job j. Sea además Tj el tiempo transcurrido desde elinicio de ejecución del job.

Entonces, el tiempo restante %(j) de la fase Map de dicho job es calculadoen este algoritmo según la ecuación 3.20, si bien en la bibliografía figuranotras técnicas distintas para estimar esta cantidad [92, 121, 122].

%(j) =

+∞ fj = 0

Tjfjpj fj 6= 0

(3.20)

Para probar este algoritmo, el equipo de Xiaotong Zhang toma como baseel código de Fair Scheduler e implementa sobre este scheduler el método deselección de tareas Reduce del algoritmo SRT.

Cuando el JobTracker es informado de que algún TaskTracker posee unslot libre para ReduceTasks, reenvía el pedido al TaskScheduler. Este com-ponente elige entre los pools de jobs de acuerdo a la lógica de Fair Scheduler.Una vez elegido el pool, se ordenan los jobs del mismo de acuerdo a su valorde %(j), en orden ascendente, y escoge el primero.

Entonces el TaskScheduler toma alguna tarea Reduce del job seleccionadoy se lo remite a su vez al JobTracker, para que finalmente sea asignado alReduceSlot libre en cuestión.

77

3.3. Trabajos relacionados

Si bien la presente tesis se aboca al estudio del problema de optimizacióndel uso de recursos en arquitecturas de procesamiento distribuido como Ma-pReduce y Hadoop, particularmente para la localidad de datos de las tareasReduce, existe un gran número de trabajos publicados donde se analizandistintos aspectos de este tipo de sistemas y buscan mejorar su performance.Algunas de estas publicaciones se mencionan brevemente a continuación.

La optimización de localidad de datos para las tareas Map recibió signi-ficativamente mayor atención desde los comienzos del modelo MapReduce.El trabajo más divulgado en sentido es el demominado Delay Scheduling,propuesto por Matei Zaharia et al. [88], donde el TaskTracker impone untiempo de espera durante el cual asigna una dada tarea Map sólo si encuen-tra un nodo-candidato que cuenten con la particón almacenada localmente, ytranscurrido ese tiempo de espera lo asigna en cualquier nodo-candidato. Talcomo fue observado en la Subsección 3.1.1, se busca favorecer la localidad dedatos de las tareas ejecutadas en su compromiso con la prioridad del ordende llegada de los jobs.

En el mismo sentido, el trabajo de Zhang et al. propone un método descheduling denominado Next-K-Node [123] en el cual se organizan los recurosa utilizar en base al cálculo de la probabilidad del tiempo que tardará cadanodo en servicio en terminar su actual MapTask, y enviar un pedido paraque se le asigne una nueva. La estimación de dicha probabilidad se realiza enbase al porcentaje de progreso actual de cada nodo en servicio (se asume quelas tareas que se encuentran más avanzadas, finalizarán antes que aquellascon menor grado progreso).

La publicación de Isard et al. desarrolla un framework denominado Quincy[124] que centra su trabajo en hallar un equilibrio en el compromiso entreoptimizar localidad de datos, y optimizar la equidad en la asignación de re-cursos para los distintos jobs. Para ello se mantienen estructuras de grafosque representan el flujo de trabajo del sistema, de las cuales se vale a la horade tomar decisiones sobre la gestión de recursos. En este caso el Schedulerfue implementado sobre Dryad [74, 125] que si bien es un sistema similar aMapReduce, presenta algunas diferencias.

Otra propuesta de esquema de asignación para optimizar la asignación deMapTasks es la propuesta por Zhenhua Guo et al. [126]. Además de contribuírcon un modelo matemático para estudiar la localidad de datos al asignar

78

tareas Map, esta publicación presenta un scheduler que aplica algoritmosdedicados a resolver el problema de asignación de sumas lineales, o LSAPpor sus siglas en inglés, para decidir a qué slot le corresponde cada tarea.

Un enfoque distinto para optimizar la localidad de datos de MapReducees el de mejorar el sistema de almacenamiento de las particiones del archivoque contiene los datos de entradas del job. Por ejemplo, Ananthanarayananet al. [127] proponen un algoritmo centralizado, llamado Scarlett, que per-mite flexibilizar el coeficiente de replicación de MapReduce (actualmente enHadoop este valor es estático, especificado por el usuario). El algoritmo to-ma la decisión de aumentar el factor de replicación para aquellos archivosdetectados como populares, al analizar las estadísticas de uso de los archivosen el cluster. De esta forma es más probable que el algoritmo de schedu-ling, independientemente de su implementación, pueda encontrar nodos enlos cuales ejecutar localmente las tareas Map.

También con el enfoque puesto en el factor de replicación de los datos,Abad et al. [52] presentan DARE, un mecanismo adaptativo y ejecutado deforma distribuida en el que participan todos los nodos del cluster. A dife-rencia de Scarlett, la implementación de DARE se enfoca en decidir en quénodos se guardarán las nuevas réplicas. Este mecanismo detecta los bloquesmás populares mediante el algoritmo ElephantTrap [128], y cada nodo re-plica ese bloque (se crea una copia local) con probabilidad p, cuyo valordebería ser preferentemente comprendido en el intervalo Θ = [0,2; 0,3] segúnlos resultados experimentales del trabajo.

Con el objetivo de evitar que las tareas Reduce de un job acapare los re-cursos (en este caso, los ReduceSlots) del cluster, Yandong Wang et al. [129]proponen un sistema de detención y re-lanzamiento de dichas tareas parasolucionar el problema de monopolización, que denominan Fair CompletionScheduler. Dado que las tareas Reduce que ya comenzaron su ejecución rea-lizaron el fetch de una porción de sus particiones correspondientes, en casode ser detenidas, resultaría poco eficiente asignarlas en su re-lanzamiento ennodos distintos (requeriría voler a movilizar esos datos fetcheados). Por lotanto, Fair Completition Scheduler intenta volver a lanzar los ReduceTasksen aquellos slots donde inicialmente fueron asignados.

Asimismo, Zhenhua Guo et al. [130] proponen implementar un sistemaautomático de ajuste de granularidad de las tareas, que decide si es necesa-rio unir un grupo de tareas evaluadas como pequeñas, o dividir una tareagrande, con el objetivo de obtener mejor balance de carga en los nodos. Sinembargo, en esta investigación está orientada a jobs que requieren intensivo

79

procesamiento sobre pequeñas cantidades de datos, por lo cual la localidadde datos no es crítica.

En el caso de Tenzing, publicado por Chattopadhyay et al. [131], se propo-ne un motor de consultas SQL diseñado para funcionar sobre una plataformaque responda al modelo MapReduce, con un mejor desempeño, eficiencia ymayor proporción de compatibilidad con el lenguaje SQL estándar que lasherramientas existentes como Pig, Hive, Flume, entre otras.

80

4

Contribución

En el presente capítulo se expone la contribución de la tesis. La Sección 4.1incluye un análisis formal del problema. En la Sección 4.2 se presenta la pro-puesta de solución, basada en el estado del arte introducido en el capítuloanterior. Finalmente, se analiza la propuesta de solución y se discuten opti-mizaciones alternativas en la Sección 4.3.

4.1. Modelización del problema

La planificación de tareas en la fase Reduce puede ser vista como unproblema de optimización, en el cual se busca asignar dichas tareas en slotsde forma tal de minimizar el tiempo requerido para finalizar el procesamientode un job o conjunto de jobs.

A continuación se expone el modelo que se utilizará para analizar la so-lución propuesta en la Sección 4.2. Puede encontrarse un resumen de lanotación en el Apéndice A. El mismo se basa en los modelos elaborados enlos trabajos desarrollados en los capítulos previos, donde se estudia la op-timización de la etapa shuffle [53, 112, 132] y otros temas similares que seencuentran relacionados y resultan de interés [126].

Una de las aristas principales de dicho problema es el estudio del costode transmisión de datos intermedios, desde los nodos que generaron estosdatos durante la ejecución de las tareas Map del job, hacia el nodo que debeejecutar la tarea Reduce.

81

4.1.1. Notación y definición de conceptos

Sea G = (V, E) la topología del cluster donde se ejecuta el sistema, com-puesta por un conjunto V de nodos y un conjunto E de conexiones entre ellos,y sea H(u,v) la función de distancia topológica entre dos nodos u,v ∈ V,tomando H(u,u) = 0 para cualquier nodo u. Hadoop asume una topologíaen árbol o jerárquica [53], como ilustra la figura 4.1.

Se asume que al transferir un conjunto de datos de tamaño X desde unnodo u a otro nodo v el costo de transferencia mediante la red es proporcionalal producto X ·H(u,v) para cualquier par de nodos u,v ∈ V con u 6= v.

El conjunto J = J1, J2, ... está compuesto por los jobs que ingresan alsistema para ser ejecutados, los cuales se encuentran indexados por orden dearribo. Cada job Ji se compone a su vez de un conjuntoMi = p1, p2, ..., pMide tareas Map y de un conjunto Ri = q1, q2, ..., qRi de tareas Reduce, ytiene asociado el momento ti de su arribo. Las cantidad Mi de tareas Mapy la cantidad Ri de tareas Reduce son específicas para cada job Ji, y vienendadas por los parámetros especificados por el usuario que lo ejecuta.

En el presente modelo, cuando se aplica sobre una tarea Map o Reduce,o sobre un conjunto de tareas, el operador | · | denota su cantidad de trabajo,medida a través del volúmen total del input. Así, el tamaño de los datos deentrada (i.e. el volumen del input) de la tarea p ∈Mi se denota mediante laexpresión |p|, mientras que el tamaño total del input del i-ésimo job es |Mi|.

De la misma forma, |q| representa el tamaño del input de la tarea q ∈ Ri,y |Ri| el tamaño total de los datos intermedios (es decir, de entrada para lastareas Reduce) del i-ésimo job.

Figura 4.1: Esquema de topología en árbol [133].

82

A su vez, se tiene en cuenta el tamaño Xpq de los resultados intermedios

producidos por la MapTask p ∈ Mi destinados a la ReduceTask q ∈ Ri, eltamaño total Xq de los datos intermedios destinados a dicha tarea Reduce,y el tamaño total Xi de los datos intermedios destinados a cualquier tareaReduce del job Ji. Las ecuaciones expresadas en 4.1 dan cuenta de la relaciónentre dichas magnitudes y el tamaño del input de cada tarea Reduce.

|q| = Xq =∑p∈Mi

Xpq ∀q ∈ Ri

|Ri| = Xi =∑q∈Ri

Xq

(4.1)

Sea ϑi : Mi → V la función que mapea cada tarea Map del job Ji al nododonde es asignada, y sea ξi : Ri → V la función que realiza el mapeo respecti-vo para las tareas Reduce del mismo job. Tomando también en consideraciónel tamaño Xq(u) del segmento de datos intermedios, almacenados en el nodou ∈ V y destinados a una dada tarea Reduce q del job Ji —que se encuentraen ejecución en el nodo ξi(q) al que fue asignada—, puede definirse entoncesel costo total para transferir los datos intermedios de aquella ReduceTaskmediante la expresión 4.2.

C(q) =∑u∈V

(Xq(u) ·H(u, ξi(q))

)∀q ∈ Ri (4.2)

Los archivos sobre los cuales se ejecutan los jobs de MapReduce se en-cuentran almacenados en el file system distribuído (e.g. HDFS), y seránrepresentados mediante el conjunto A = A1, A2, .... Cada archivo As estácompuesto por un conjunto Bs = Bs

1, Bs2, ..., B

sl de l bloques o particio-

nes. Como fue expresado en la Subsección 2.4.1, el factor de replicación ρrepresenta la cantidad de nodos del cluster que almacenan una copia de cadabloque.

Además, mediante la función πs : Bs → Vρ se designa al mapeo de cadabloque b ∈ Bs a un conjunto πs(b) compuesto por ρ nodos distintos delcluster. Por lo tanto, dado un job Ji que es ejecutado sobre un archivo As,y dada una tarea Map p ∈ Mi de ese job que debe procesar cierto bloqueb ∈ Bs de aquel archivo, se tiene que dicha tarea se ejecuta con localidad dedatos si y sólo si se cumple que ϑi(p) ∈ πs(b).

83

Cada nodo u del cluster cuenta asimismo con un conjunto Sm(u) de slotspara tareas Map, y con un conjunto Sr(u) de slots para tareas Reduce. Losconjuntos Sm y Sr, definidos en 4.3, denotan a su vez el total de slots delcluster de cada tipo.

Sm =⋃u∈VSm(u)

Sr =⋃u∈VSr(u)

(4.3)

En el instante de tiempo t, cada uno de los slots de ambos tipos puedeencontrarse disponible o albergando la ejecución de una tarea. Se denotamediante el operador Dt(·), aplicado sobre cualquiera de los conjuntos deslots descriptos, al subconjunto formado por los slots que en aquel instantese encuentran disponibles.

El scheduler es representado en base al trabajo de Jian Tan et al. [100].El ingreso de jobs al sistema se modela mediante un proceso de Poisson,donde el intervalo de tiempo Ii = ti − ti−1 entre los arribos de dos jobsconsecutivos Ji−1 e Ji se encuentra distribuído de forma exponencial contasa λ. La cantidad máxima de jobs simultáneamente en servicio es K.

Ante el arribo del i-ésimo job, se lo une a una cola de espera en el casode haber K jobs en servicio, o bien se inicia su ejecución en caso contrario.Al iniciarse su ejecución, las tareas del conjunto Mi son incorporadas a lacola de tareas Map, que será modelizada mediante la ya mencionada colaM/GI/1/PS (processor-sharing queue). Asimismo, la cola de tareas Reducees representada mediante la notación de Kendall según M/GI/ς, tomandoς = ||Sr||, esto es, la cantidad de slots para ReduceTasks en el cluster.

A diferencia del caso de las tareas Map, que ante la entrada en servicio deljob se incluyen a la cola de tareas Map inmediatamente, las tareas Reducese van incorporando a la cola correspondiente a medida que el schedulerasí lo disponga. Se utilizará ψ : [0, 1] → [0, 1] para hacer referencia a laproporción de tareas de Ri incorporadas a la cola Reduce en función delprogreso (la proporción de tareas completadas) de la fase Map de cada jobJi. Por ejemplo, en caso de utilizarse el mecanismo de early shuffle descriptoen la Subsección 2.4.3, se tendría ψ(x) = Θ(x − a), es decir un escalón deHeaviside especificando la proporción a ∈ [0, 1] de tareas Map completas.Para el caso de Coupling Scheduler [106] se tendría en cambio ψ(x) = x.

84

Se utilizará xi para hacer referencia al progreso de la fase Map del job Jien un momento dado, medido como la proporción de tareas Map completasrespecto al total de tareas Map de dicho job. Asimismo, yi será utilizadapara referir a la proporción de tareas Reduce de un job Ji que se encuentrancompletas, en ejecución, o bien incorporadas a la cola de espera de tareasReduce, respecto al total de tareas Reduce de ese mismo job.

El tamaño parcial de los datos intermedios, destinados a una tarea Re-duce q ∈ Ri y almacenados en un nodo u ∈ V, que ya fueron generadosen un dado instante t, será simbolizado mediante Xq(u). El tamaño par-cial de los mismos datos, almacenados en cualquier nodo, serán denotadosmediante Xq. De igual manera, Xi(u) y Xi harán referencia al tamaño par-cial de datos intermedios ya generados, destinados a cualquier tarea reducedel job Ji, almacenenados en el nodo u y almacenados en cualquier nodo,respectivamente.

La duración de cada tarea Map depende tanto de la función map() definidapor el usuario para el job, de los datos contenidos en el bloque asignado adicha tarea, de la capacidad de procesamiento del nodo al cual se asigna, yde la localidad de datos.

Por lo tanto, aún si puede considerarse uniforme la capacidad de proce-samiento de todos los nodos del cluster, dicha duración depende también delesquema de asignación ϑi, en tanto que las tareas asignadas a nodos dondese ejecutarán con localidad de datos se ejecutarían en menor tiempo.

En el caso de las tareas Reduce, su duración depende también de la fun-ción reduce() propia del job, del tamaño y la naturaleza de los datos in-termedios que le corresponden como input y del nodo al que fue asignada.Pero depende además de la ubicación de las tareas Map del mismo job, entanto que afectan al costo de transferencia de datos intermedios, como fueexpresado en la ecuación 4.2. Es por ello que la duración de las tareas Reducese ve afectada tanto por el esquema de asignación de tareas Map ϑi, comopor el esquema de asignación de tareas Reduce ξi.

Dadas ϑi y ξi, se representa mediante Ti la duración del job Ji, es decir elintervalo de tiempo trasncurrido desde el momento en que entra en serviciohasta que finaliza su ejecución.

85

4.1.2. Determinación del problema

Uno de los aspectos principales de la administración eficiente de recursosen MapReduce es la optimización del componente de scheduling. El mismoactúa como coordinador de los recursos del cluster, en tanto que cumplela función de asignar recursos del cluster para las tareas Map y Reducependientes de los jobs en ejecución. En particular, la localidad de datosrepresenta uno de los factores más decisivos para la performance del sistema.

Para las tareas Map, optimizar su localidad de datos implica maximizar laprobabilidad de que cada una de ellas será ejecutada en alguno de los nodosque almacena una copia del bloque que debe procesar. Las MapTasks de unmismo job son independientes entre sí, y cada una de ellas ejecuta un bloquede datos de entrada (de tamaño fijo), con lo cual resulta relativamente directala implementación de dicha optimización. Es por ello que la localidad de datospara las tareas Map se encuentra ampliamente estudiada y optimizada en lamayor parte de las implementaciones del Scheduler del sistema.

La localidad de datos de las tareas Reduce representa un desafío mayor,dada la complejidad introducida por las dependencias de la etapa shuffle. Porlo tanto, en la mayoría de las investigaciones orientadas a la optimización deMapReduce no se tiene en cuenta que los resultados intermedios necesitan sertransportados hacia los nodos que ejecutan las ReduceTasks. No obstante,esta situación se encuentra analizada en profundidad por las investigacionescitadas en el Capítulo 3.

El presente trabajo aborda entonces el problema de encontrar un algorit-mo de scheduling que (a) incorpore tareas Reduce a la cola correspondienteen base al progreso de la fase Map de acuerdo a una función ψ, (b) admi-nistre de forma eficiente los recursos de la plataforma, evitando inurrir ensubutilización, y (c) obtenga, para cada job Ji, un esquema de asignación ξide tareas Reduce, de forma tal que minimice la expresión 4.4.

E

∑q∈Ri

∑u∈V

(Xq(u) ·H(u, ξi(q))

) (4.4)

En tanto que la etapa shuffle/sort representa un cuello de botella parauna cantidad significativa de ejecuciónes de jobs de MapReduce, en estoscasos dicha minimización, en conjunto con un uso eficiente de los recursosdel cluster, derivaría también en la reducción de Ti.

86

4.2. Solución propuesta

La contribución principal de la presente tesis consiste en la elaboracióndel algoritmo Flexible Coupling Scheduler. Este esquema de schedulingde MapReduce, basado en el trabajo de Jian Tan et al. [98], busca flexibili-zar la asignación de tareas Reduce de Coupling Scheduler para aumentar laeficiencia en el uso de recursos y al mismo tiempo obtener un buen nivel delocalidad de datos en la etapa shuffle, minimizando así la carga de transmi-sión de datos en el cluster. De esta forma se pretende mejorar el tiempo deprocesamiento de los jobs. Flexible Coupling Scheduler implementa un me-canismo similar a delay scheduling [88] para el caso de asignación de tareasReduce, en el contexto del lanzamiento gradual de dichas tareas utilizadopor Coupling Scheduler.

En situaciones donde existen dos o más jobs con ReduceTasks esperandoa ser ejecutadas, Coupling Scheduler prioriza los jobs con mayor mismatch(desbalance entre el progreso de tareas Map y Reduce) para evaluar a cuálde ellos asignarle un nuevo ReduceSlot libre. Mediante el algoritmo WaitScheduling, se le asigna al job seleccionado un slot en el nodo que minimiza ladistancia a sus datos intermedios, entre aquellos nodos con slots disponibles.No obstante, para ello puede dejar pasar hasta 3N heartbeats sin realizarninguna asignación, siendo N la cantidad de nodos del cluster.

El esquema propuesto en este trabajo mantiene la manera de calcular elmismatch de Coupling Scheduler y el algoritmo Wait Scheduling, pero realizauna asignación extra tras haber salteado N heartbeats. Se utilizará el costopara transferir datos intermedios, presentado en la sección anterior mediantela ecuación 4.2, para comparar el nivel de aprovechamiento que pueden hacerdistintos jobs de un slot ubicado en un nodo u ∈ V.

Enunciado de la tesis. Un esquema de scheduling de MapReduce basado enCoupling Scheduler, que incorpore un mecanismo similar a delay schedulingpero aplicado a la asignación de tareas Reduce, puede mejorar la performancedel sistema en tanto que mantiene un buen nivel de localidad de datos y almismo tiempo evita incurrir en subutilización de recursos.

Flexible Coupling Scheduler, el scheduler propuesto en este trabajo, se-rá desarrollado en mayor detalle a continuación. En la Subsección 4.2.1 sedescribe el algoritmo de asignación de tareas Reduce utilizado en dicho sche-duler, y en la Subsección 4.2.2 se presenta su diseño.

87

4.2.1. Descripción del algoritmo

El esquema de lanzamiento gradual de tareas Reduce implementado enCoupling Scheduler, que se detalla en la Subsección 3.1.7, resulta de graninterés dado que posibilita la optimización simultánea de los esquemas deasignación ϑi y ξi para cada job Ji, conforme evoluciona su ejecución. Sibien Coupling Scheduler fue diseñado con esta característica como uno desus objetivos, aún existe espacio para introducir mejoras.

El algoritmo de asignación de Coupling Scheduler incorpora nuevas tareasa la cola de tareas Reduce en base al progreso de la fase Map, de acuerdoa una función ψ. Específicamente, se incorporarán nuevas tareas Reduce deun job Ji a la cola si y sólo si se cumple ψ(xi) > yi, donde xi hace referenciaa la proporción de tareas Map completas respecto del total de ese job, e yies la proporción de ReduceTasks que se encuentran completas, en ejecucióno en espera en la cola, respecto del total de tareas Reduce del job.

Cada vez que el JobTracker recibe un heartbeat, en caso de no poseerningún job seleccionado, se establece aquel job con mayor mismatch comocandidato a tomar un nuevo ReduceSlot y se setea en cero un contador quetraquea los heartbeats transcurridos. De lo contrario, en caso de ya tenerun job candidato, se incrementa el contador de espera y el algoritmo WaitScheduling [98] evalúa lanzar una tarea en el nodo que envió el heartbeat.

Si el nodo examinado posee algún slot libre y el contador tiene un valorentre 0 y 3N , entonces se compara el nodo que envió el heartbeat con los no-dos almacenados en los primeros lugares de la lista L(i) definida en la secciónSubsección 3.1.7 y asigna una tarea Reduce al nodo en caso de coincidir. Encambio, si el contador tiene un valor mayor a 3N , entonces asigna sin con-diciones una ReduceTask del job candidato al nodo examinado. Finalmente,de haber lanzado alguna tarea, se asigna un valor vacío al job candidato,para ser elegido en la siguiente iteración.

El esquema de asignación de tareas Reduce propuesto como contribuciónde la tesis introduce un mecanismo alternativo de selección ante la llegadade un nuevo ReduceSlot libre, en caso de que haya más de un job con tareasReduce en la cola. Cuando al job Ji1 con mayor mismatch le correspondeun nuevo slot libre, pero han transurrido N heartbeats sin asignarse Redu-ceTasks y existe algún otro job Ji2 con tareas Reduce en la cola de esperaque podría ejecutarlas con mejor localidad de datos, entonces puede cederlopara que la tarea de Ji2 se ejecute optimizando dicha localidad de datos.

88

4.2.2. Diseño de la implementación

Para implementar el algoritmo presentado en la subsección anterior, re-sulta necesario contar con un operador Ω(v, Ji) que estime, para un dadojob Ji y un dado nodo v ∈ V, el costo de transferencia de todos los datos in-termedios producidos por las tareas Map de dicho job al nodo v, teniendo encuenta el tamaño de dichos datos intermedios, y de la topología del cluster.

Dicha estimación puede realizarse en base a la ecuación 4.2, suponiendoque una nueva tarea Reduce será ejecutada en un slot del nodo v. En estesentido se toma en cuenta el tamaño κ(u) de los datos intermedios destinadosa cualquier tarea Reduce del job Ji, almacenados en cada nodo u ∈ V, quese espera que sean producidos al finalizar la fase Map. La manera de calcularel estimador Ω(v, Ji) queda expresada mediante la ecuación 4.5.

Ω(v, Ji) =1

γ·∑u∈V

(κ(u) ·H(u,v)

)(4.5)

La sumatoria que figura en dicha expresión calcula la distancia total pon-derada (similar a la planteada por Hammoud et al. [112] en la ecuación 3.8,utilizada para CoGRS ) tomando en cuenta una estimación κ(u) del tama-ño de datos intermedios almacenados en cada nodo u, dado que el tamañopreciso Xi(u) de resultados intermedios no se conoce aún. El resultado dela sumatoria es dividido por el cuadrado de la cantidad total de datos in-termedios estimados de ese job, γ. Con ello se evita priorizar aquellos jobscuyo volúmen estimado de resultados intermedios es menor. Las ecuaciones4.6 determinan la manera de calcular κ(u) y γ.

κ(u) =Xi(u)

xi ·Rixi > 0

γ =

(∑u∈V

κ(u)

)2 (4.6)

El cálculo de κ(u) toma la cantidad Xi(u) de datos intermedios del job Jigenerados hasta ese momento y almacenados en el nodo u ∈ V, y los dividepor el progreso xi de la fase Map y por la cantidad de tareas Reduce del job,Ri, a fin de obtener la estimación de Xq(u).

89

Algoritmo 2 Asignación de ReduceTasks en Flexible Coupling SchedulerValores estáticos: heartbeats ← 0, umin ← Null, Jmin ← Null y

Ωmin ←∞.Input: El nodo u ∈ V que declara un nuevo ReduceSlot libre y la cola de

tareas Reduce pendientes para ser asignadas Q.Output: Computa la asignación de tareas Reduce para el nodo u.1: heartbeats← heartbeats+ 12: if ReduceSlotLibres(u) and not Vacía(Q) then3: tareaAsignada← WaitScheduling(u)4: if not tareaAsignada then5: ActualizarEstimaciones(u, Q)6: end if7: end if8: if heartbeats > N and Jmin 6= Null then9: AsignarReduceTask(umin, Jmin, Q)

10: ResetearEstimaciones11: end if

Los detalles de la asignación de tareas Reduce según el esquema propuestoen Flexible Coupling Scheduler pueden ser encontrados en el Algoritmo 2 yserán descriptos a continuación.

La función ReduceSlotLibres utilizada en la condición de la línea 2 re-torna True en caso de que el nodo pasado como parámetro posea al menosun ReduceSlot libre, y False en caso contrario. Asimismo, la función Vacíainvocada en la misma línea se evalúa en True si la estructura de datos a laque se aplica (en este caso, una cola) no posee ningún elemento, y False encualquier otro.

La línea 3 ejecuta la lógica del algoritmo Wait Scheduling. No obstante,Flexible Coupling Scheduler evalúa la posibilidad de lanzar tareas Reduce dejobs que no necesariamente se encuentran primeros en el ranking de valor demismatch, en caso de que para el nodo que se está evaluando no se hubieralanzado ninguna tarea Reduce. Se requiere entonces adaptar el algoritmoWait Scheduling para retornar información sobre si pudo o no asignar unatarea Reduce en el nodo evaluado. Dicha adaptación figura en el Algorit-mo 3. Obsérvese además que, a diferencia de la implementación de CouplingScheduler [98], no se saltea el heartbeat en caso de tener que computarse eljob de mayor mismatch (lo cual aporta una pequeña optimización adicionalpara la contribución de la presente tesis).

90

Algoritmo 3 Wait Scheduling adaptado para Flexible Coupling Scheduler

Valores estáticos: candidato← Null y m0 ← 0.Input: El nodo u ∈ V que declara un nuevo ReduceSlot libre y la cola de

tareas Reduce pendientes para ser asignadas Q.Output: True si se asignó alguna tarea, y False en caso contrario.1: if candidato = Null then2: (J,m0)← EvaluarCandidatos3: if m0 > 0 then4: candidato← J5: end if6: end if7: wait← wait+ 18: i← bwait/Nc9: if i < 3 then10: if u in Li(J) and not EjecutaReducer(de=J , en=u) then11: exito← AsignarReduceTask(u, J , Q)12: if exito then13: candidato← Null14: wait← 015: return True16: else17: wait← i ·N + 118: end if19: end if20: else21: exito← AsignarReduceTask(u, J , Q)22: if exito then23: candidato← Null24: wait← 025: return True26: end if27: end if28: return False

91

El procedimiento ActualizarEstimaciones, ejecutado en la línea 5 delAlgoritmo 2, procesa los heartbeats que el algoritmo Wait Scheduling noutilizó para asignar tareas Reduce. En estos casos, se toma el nodo u queenvió el heartbeat y se compara (mediante el estimador Ω) la posibilidad delanzar ReduceTasks de cada uno de los jobs con tareas en la cola Q de tareasReduce pendientes.

Su definición queda expresada, a su vez, mediante el Algoritmo 4. Losparámetros que recibe entonces el procedimiento son el nodo u y la cola Q.En la línea 2 se genera, a partir de Q, una lista L con todos los jobs con almenos una tarea Reduce pendiente a ser ejecutada. Para cada job Ji de dichalista, se calcula el valor del estimador Ω(u, Ji) en la línea 4. Las variablesestáticas umin, Jmin y Ωmin almacenan la información sobre el nodo, dondede ser asignada una tarea Reduce de un dado job, minimiza el valor delestimador.

Durante cada uno de los heartbeats rechazados por Wait Scheduling, seevalúa el estimador Ω y se actualizan las estimaciones en caso de ser ne-cesario. Así, en caso de que para un job Ji el valor del estimador Ω(u, Ji)sea menor al mínimo actual Ωmin, entonces se actualizan las tres variablesestáticas mencionadas.

Es interesante notar que el cálculo de estos estimadores (y la lógica quetermina lanzando tareas Reduce en base a dichos resultados) puede ser pro-cesado en paralelo, en algún otro nodo que comparta ciertos elementos deinformación con el JobTracker. De esta forma es posible mantener al JobTrac-ker con una carga de procesamiento similar a la que posee bajo el esquemade Coupling Scheduler, evitando sobrecargarlo.

En la línea 9 del Algoritmo 2 se intenta asignar en algún slot libre delnodo umin alguna de las tareas del job Jmin que se encuentran en la cola Qde espera de ReduceTasks a ser ejecutadas. La función AsignarReduceTaskrecibe estos elementos como parámetros, y devuelve True en caso de teneréxito en el intento, y False si no se pudo asignar.

Cabe notar que existen diversos motivos por los cuales podría ocurrirque fracase el intento de asignar alguna tarea Reduce de dicho job en aquelnodo. Por ejemplo, como se contempla también en el trabajo de Jian Tanet al. [98], existe la posibilidad de que el nodo al cual se quiere asignar unanueva tarea Reduce posea slots libres, pero por sobrecarga de algún recursonecesario (e.g. memoria) para ejecutarla correctamente, el TaskTracker enejecución en aquel nodo termine rechazando la tarea.

92

Otra posibilidad por la cual podría fracasar la asignación es que el jobJmin sea el mismo que el tomado como candidato por Wait Scheduling, esdecir el de mayor mismatch. En caso de que Wait Scheduling asignara todaslas tareas Reduce pendientes de Jmin, no quedarían tareas Reduce de dichojob en Q.

En cualquier caso, sea que tuvo éxito o no, se resetea la información delnodo umin ∈ V y el job Jmin que minimizan valor del estimador Ω, dadoque en caso de ocurrir algún evento que haga fracasar constantemente elintento de asignación, entonces dicha información nunca se actualizaría yel algoritmo de asignación de tareas Reduce pasaría a ser el mismo que elpropuesto en Coupling Scheduler.

Por otro lado, el procedimiento ResetearEstimaciones ejecutado en lalínea 10 del Algoritmo 2 le devuelve el valor inicial (especificados en la de-finición de valores estáticos del algoritmo) al contador heartbeats y a lasvariables que almacenan el nodo umin, el job Jmin y el valor Ωmin que co-rresponden la mejor posibilidad de asignación entre aquellos jobs con tareasReduce por asignar y nodos con slots libres.

Algoritmo 4 Actualización de valores estáticos en base al estimador Ω

Valores estáticos: umin ← Null, Jmin ← Null y Ωmin ←∞.Input: El nodo u ∈ V que declara un nuevo ReduceSlot libre y la cola de

tareas Reduce pendientes para ser asignadas Q.Output: Actualización de los valores estáticos.1: procedure ActualizarEstimaciones(u, Q)2: L ← Jobs(Q)3: for each Ji in L do4: Ωi ← Ω(u, Ji)5: if Ωi < Ωmin then6: umin ← u7: Jmin ← Ji8: Ωmin ← Ωi

9: end if10: end for11: end procedure

93

4.3. Discusión

En el presente capítulo se desarrolló el algoritmo de asignación de tareasReduce que forma parte de la contribución de la tesis, denominado FlexibleCoupling Scheduler. Introduciendo un mecanismo similar a delay schedulingal esquema de Coupling Scheduler desarrollado por Jian Tan et al. [98], con elfin de flexibilizar su sistema de selección de jobs para asignar los ReduceSlotslibres, se logra mejorar la localidad de datos de la fase Reduce, y se buscade esta forma optimizar la performance de la plataforma.

Resulta interesante analizar entonces algunos aspectos del algoritmo pre-sentado. En particular, la ecuación 4.5 es utilizada para obtener la estimaciónde carga de tráfico en el cluster provocada al asignar una tarea Reduce deljob Ji a alguno de los slots de un dado nodo u ∈ V. El algoritmo emplea estaestimación para comparar el nivel de localidad de datos que se obtendría encaso de asignarse tareas Reduce que pertenecen a distintos jobs.

Al momento de calcular dicha carga de tráfico, en el caso general se cuentasólo con información parcial del volúmen y de la ubicación de los datosintermedios destinados a cada ReduceTask del job.

El cálculo de la estimación debe tener en cuenta el progreso de la faseMap de cada job. De lo contrario, aquellos jobs más avanzados (con mayorproporción de resultados intermedios ya generados) tenderían a tener mayorprioridad a la hora de repartir los slots libres, vulnerando el sistema demismatch que implementa Coupling Scheduler.

Asumiendo, como en el análisis de Jian Tan et al. [53], un progreso linealdel crecimiento de los resultados intermedios, se calcula κ(u) como el cocienteentre la cantidad (observada en ese instante) de resultados intermediosXi(u)destinados a cualquier tarea Reduce del job y el producto entre la proporciónxi de tareas Map completas, con lo cual se obtiene una estimación de Xi(u).Además, se divide al resultado por la cantidad Ri de tareas reduce del job,para obtener así un estimado del volúmen promedio de resultados intermediospara cada tarea Reduce del job, según lo expresado en 4.6.

A partir de la ecuación 4.5 es posible además notar que la distancia totalponderada es dividida por γ, es decir el cuadrado de la suma de la cantidad(observada en ese instante) de resultados intermedios almacenados en cadanodo. Por lo tanto, entre dos jobs en condiciones idénticas excepto por lacantidad total de datos intermedios, el job con mayor volúmen de datosintermedios obtendrá un valor menor para la estimación, y con ello mayor

94

prioridad para recibir recursos durante la fase shuffle. Se busca así mejorarcon mayor probabilidad la localidad de datos para las tareas Reduce de losjobs con mayor volúmen de resultados intermedios.

Por otro lado, cabe observar también que el estimador Ω(v, Ji) asume quela distribución de resultados intermedios destinadas a cada tarea Reduce deljob se encuentra balanceada.

Una versión alternativa al algoritmo presentado en la sección anteriorpodría entonces utilizar estimaciones más específicas para evitar realizar talhipótesis, como por ejemplo calcular mediante la ecuación 4.7 la localidadde datos estimada Ωq(v) para cada tarea Reduce q ∈ Ri, en lugar de laestimación Ω(v, Ji) general del job Ji para una tarea Reduce promedio.

Ωq(v) =1

γq·∑u∈V

(κq(u) ·H(u,v)

)(4.7)

El operador Ωq(v) realiza entonces una estimación de C(q) en caso deque ξi(q) = u, según se define en la ecuación 4.2, y luego la divide por lacantidad γq. Para ello, κq(u) y γq deberían ser calculadas a partir de lacantidad observada Xq(u) de resultados intermedios para una dada tareaReduce q ∈ Ri en particular. Estas magnitudes quedan expresadas en lasecuaciónes 4.8.

κq(u) =Xq(u)

xixi > 0

γq =

(∑u∈V

κq(u)

)2 (4.8)

En este caso, en la línea 4 del Algoritmo 4 se buscaría, para cada job Ji,la tarea r ∈ Ri cuyo valor de Ωr(v) sea el máximo. Esa tarea representaríaentonces al job en lo que sigue del algoritmo, y de ser seleccionado para recibirel slot libre, la tarea r sería asignada al slot. Para ello resultaría entoncesconveniente guardar, en lugar de una cola Q de tareas Reduce específicaspendientes a ser ejecutadas, la cantidad de tareas Reduce pendientes porcada job en ejecución, de modo de recorrer todas las tareas Reduce quetodavía no se encuentran en ejecución

95

Algoritmo 5 Alternativa para la actualización de valores estáticosValores estáticos: umin ← Null, qmin ← Null y Ωmin ←∞.Input: El nodo u ∈ V que declara un nuevo ReduceSlot libre.Output: Actualización de los valores estáticos.1: procedure ActualizarEstimaciones(u)2: L ← JobsEnEjecuciónConMismatchPositivo3: for each Ji in L do4: for each q in ReduceTasksPendientes(Ji) do5: Ωi ← Ωq(u)6: if Ωi < Ωmin then7: umin ← u8: qmin ← q9: Ωmin ← Ωi

10: end if11: end for12: end for13: end procedure

La actualización de dicho algoritmo, adaptado para la alternativa plan-teada, podría ser aquella definida en el Algoritmo 5. En este caso no se recibepor parámetro la cola de tareas Reduce pendientes Q, dado que no existiría,y en su lugar se cuenta con un sistema que traquea cuantas ReduceTasks lecorresponde ejecutar en un dado instante a cada job Ji.

La función JobsEnEjecuciónConMismatchPositivo que es ejecutada enla línea 2 obtiene todos los jobs con derecho a ejecutar en este instante almenos una tarea Reduce, de acuerdo a la función ψ y a su progreso de la faseMap. Para cada uno de estos jobs, se realiza entonces la misma actualizaciónde los valores estáticos que en la versión presentada en la Subsección 4.2.2,pero en este caso con un nivel de granularidad más fina.

En la línea 4, se obtienen mediante la función ReduceTasksPendientesaquellas tareas Reduce, pertenecientes al job Ji evaluado, que todavía nose encuentran en ejecución. Para cada una de ellas se computa Ωq(u) en lalínea 5, y en caso de que sea menor al mínimo actual, se actualizan los valoresestáticos correspondientes.

96

5

Resultados experimentales

Con el fin de evaluar y validar las ideas planteadas en el capítulo anterior,se llevó adelante una serie de simulaciones experimentales del proceso descheduling en Hadoop, implementación de código abierto de MapReduce, ycuyos resultados serán abordados a continuación.

El presente capítulo se encuentra estructurado en tres secciones. En pri-mer lugar, la Sección 5.1 describe los criterios que serán tomados en cuenta ala hora de evaluar Flexible Coupling Scheduler, introducido como contribu-ción principal del trabajo, y compararlo con otros schedulers. La Sección 5.2presenta los resultados experimentales obtenidos mediante simulaciones dela plataforma de MapReduce, con el objetivo de evaluar las propuestas pre-sentadas en el Capítulo 4 y proporcionar una validación de lo establecido.Finalmente se expone un análisis de dichos resultados en la Sección 5.3.

5.1. Métricas a tener en cuenta

Como fue mencionado en la Subsección 2.4.3, para poder comparar la per-formance de distintos algoritmos (en este caso, de schedulers de MapReduce)es necesario definir métricas. Durante la experimentación llevada a cabo en elpresente trabajo, se tuvieron en cuenta el flowtime y el makespan [67, 68, 69]como métricas principales. Adicionalmente serán tomados en cuenta el nivelde uso de la red en cada instante de tiempo, métrica que en la bibliografíafigura como network load, y el nivel de localidad de datos obtenido por lastareas Reduce tras ser asignadas.

97

A continuación se describen en mayor detalle cada una de las métricasconsideradas para analizar los resultados experimentales presentados en estecapítulo, teniendo en cuenta lo establecido en Subsección 4.1.2.

Flowtime es la métrica que hace referencia al período requerido paracompletar un determinado job, es decir, la sumatoria de tiempo trans-currido entre el arribo de cada job hasta la finalización del mismo. Enel contexto del presente trabajo, la optimización del flowtime equivalea minimizar el tiempo promedio de finalización de los jobs.

El makespan resulta de medir el tiempo transcurrido entre el arribodel primer job hasta el momento en el que se completa el último job.Esto es, puesto en términos de la experimentación desarrollada en elpresente capítulo, el tiempo total que toma un algoritmo de schedulingen procesar y completar una cantidad dada de jobs en conjunto.

Las mediciones del uso de la red en cada instante de tiempo dan cuen-ta del volúmen de datos que son transportados en un momento dadoen el cluster. Esta métrica resulta importante en tanto que, como re-curso limitado, una red sobrecargada podría degradar ampliamente laperformance del algoritmo de scheduling.

La localidad de datos como métrica, en el presente contexto, hace re-ferencia a la ocupación de red producto de la cantidad de resultadosintermedios que se deben transportar hacia los nodos que ejecutan lostareas Reduce, por lo cual se encuentra ampliamente relacionada conel ítem anterior.

Durante la simulación, se evaluarán las mencionadas métricas en diversosexperimentos para obtener un panorama general de la performance de cadauno de los algoritmos de scheduling. Los elementos tomados como variablesserán la distribución de los tamaños de archivos que funcionarán como inputde los jobs, el momento en el que cada job arriba al sistema, y la cantidadde nodos que forman el cluster.

En los casos donde en la experimentación intervienen jobs con tamañosde input variados, serán observadas además las características del flowtimeespecífico para cada grupo de jobs, agrupados según su tamaño de input. Deesta forma es posible evaluar la performance de cada algoritmo respecto acada tipo de job, y evaluar si bajo alguno de los esquemas se perjudican ono los jobs más chicos en deterimento de los más grandes.

98

5.2. Experimentación

La presente sección expone la metodología utilizada para llevar a cabo lavalidación experimental de la contribución de la tesis en la Subsección 5.2.1.La sección Subsección 5.2.2 detalla el diseño del simulador implementadopara realizar dicha experimentación, y en la Subsección 5.2.3 se presentanlos resultados obtenidos.

5.2.1. Metodología

Para evaluar el esquema de Flexible Coupling Scheduler, se desarrollóun simulador denominado mrschedulersim. El mismo fue implementado enPython y tiene en cuenta aquellos los aspectos del proceso de administraciónde recursos de MapReduce mencionados en la Sección 4.1, relativos a lasdecisiones tomadas por el componente de scheduling de la plataforma.

La razón por la cual se desarrolló y utilizó un simulador del procesode scheduling, en lugar de llevar adelante la implementación de FlexibleCoupling Scheduler en el código de Hadoop para evaluarlo directamente através de experimentaciones con la propia plataforma, es el elevado nivel decomplejidad para implementar dicho scheduler como plugging de Hadoop.

Si bien se trata de un desafío interesante, el autor del presente trabajoconsidera que su implementación se encuentra fuera del alcance de una tesisde grado. En particular, considerando que código fuente del esquema sobre elcual se basa la contribución, es decir Coupling Scheduler [98], no es público.Adicionalmente, su correcta validadación [134] implicaría requerimientos deinfraestructura a la que no se dispone.

A su vez, se decidió implementar un nuevo simulador en lugar de utilizaralguno de los ya existentes, dado que éstos no se ajustan a las necesidadesdel presente trabajo y por lo tanto no brindan la posibilidad de validar lomencionado en el capítulo anterior.

Un análisis de los simuladores existentes para Hadoop y MapReduce re-vela que los mismos están orientados a predecir la performance de la herra-mienta en tanto que la misma se encuentra en ejecución, y utiliza el códigofuente de Hadoop, o bien no tienen en cuenta elementos importantes del pro-ceso de scheduling que resultan imprescindibles para una correcta evaluaciónde la contribución.

99

Un proyecto que apunta a simular la performance de Hadoop, y en estecaso en particular la toma de decisiones del scheduler, es aquel desarrolladopor Arun C. Murthy, denominado Mumak [135]. No obstante, este simuladorutiliza el código fuente del scheduler evaluado para producir sus resultados,por lo cual requiere que el mismo se encuentre implementado.

SimMR de Verma et al. [136], y el simulador presentado en la publica-ción de Cardona et al. [137] son otros dos proyectos que buscan predecirla performance del sistema a partir de simulaciones del proceso involucradoen la ejecución de jobs de MapReduce. Estos simuladores, sin embargo, nocontemplan partes importantes de tal proceso.

MRPerf, publicado por Wang et al. [138, 139], es un simulador más com-pleto, pero que también utiliza el código fuente de Hadoop, y en particulardel scheduler, para predecir la performance de la plataforma.

Asimismo, existen otros simuladores, en particular SimMapReduce [140],HSim [141] y MRSim [142], que están implementados con el objetivo depredecir la performance de aplicaciones donde se ejecuta, de forma individual,un sólo job [139], y por lo tanto no es posible evaluar cómo se desenvuelveel algoritmo de scheduling ante la presencia de múltiples jobs ejecutándose.

Para comparar su perfrmance con la de Flexible Coupling Scheduler, seimplementaron en la simulación desarrollada otros dos schedulers: Fair Sche-duler, dado que es el esquema más utilizado en la actualidad, y CouplingScheduler dado que es aquel en el que está basada la contribución de la tesis.

En cuanto a los experimentos, se determinaron tres situaciones a ser eva-luadas. Todas las situaciones evaluadas fueron ejecutadas con el simuladorsucesivas veces, y los resultados presentados a continuación, para cada ex-perimento, son tomados como el promedio de lo obtenido por el programa.De esta forma se minimiza el impacto de elementos aleatoreos utilizadospor el simulador (que son necesarios para representar el funcionamiento deMapReduce en entornos productivos).

El primer esquema consiste en la ejecución de múltiples jobs con distintostamaños de input, cuya distribución de tiempos de arribos respeta una dis-tribución exponencial, compartiendo y compitiendo por los recursos de uncluster de 66 nodos (valor típico en instalaciones industriales [143]). De estaforma se busca capturar la performance de los schedulers en circunstanciassimilares al uso que se le da a la plataforma en ambientes de producción,para compararlos a partir de las métricas establecidas en la sección anterior.

100

Un segundo esquema consta de ejecuciones similares a la mencionada,también con tamaños de input aleatorios y arribos distribuídos de formaexponencial, pero variando (a) la cantidad N de nodos que componen elcluster, y (b) la tasa λ de arribos de jobs al sistema. Con ello se pretendeobservar el impacto de aquellos elementos sobre la performance de cada unode los schedulers.

5.2.2. Diseño

El simulador mrschedulersim fue desarrollado con el objetivo de validarla contribución de la tesis. Implementado en Python, su código fuente puedeser encontrado en el Apéndice B. Mediante una serie de clases que modelanlos elementos y aspectos relevantes que intervienen en el proceso de schedu-ling de MapReduce, el diseño de este simulador se realizó teniendo en cuentalo observado en la Sección 4.1 del capítulo anterior.

Además, para contemplar las observaciones publicadas en el trabajo deSoila Kavulya et al. [109] respecto de las propiedades estadísticas de losjobs ejecutados en entornos de producción de MapReduce, mrschedulersimposiciona de forma aleatoria las réplicas de los archivos en los nodos delcluster, y genera jobs a partir de determinadas distribuciones, especificadasen dicha publicación. Para ello, en la simulación intervienen una serie deelementos aleatorios. El arribo de los jobs al sistema está modelado a partir deuna distribución exponencial, como fue establecido en el modelo presentadoen la Sección 4.1. El tamaño del input de cada uno de los jobs es escogidode forma aleatoria de entre un conjunto de posibles tamaños. Los heartbeatsde cada nodo llegan, en cada ronda, en orden aleatorio.

Asimismo, las réplicas de cada split de todos los archivos almacenadosen el filesystem se distribuyen de manera aleatoria. La duración de las ta-reas Reduce y de la etapa de procesamiento de las tareas Reduce se mo-delan mediante distribuciones normales con media y varianza tomadas delas mencionadas propiedades estadísticas. Así, los jobs cuyas tareas tienenmayor duración representan a aquellos con mayores requerimientos de proce-samiento. De igual manera, la proporción de resultados intermedios respectoal input de las tareas Map, y la proporción de tareas Reduce de job respectodel número de tareas Map de cada job se calcula de forma aleatoria. Los jobscon mayor proporción de archivos intermedios, por lo tanto, representan aaquellos de gran carga de I/O.

101

La clase Cluster gestiona todos los aspectos relativos a la infraestruc-tura de la plataforma. Para ello hace uso de las clases Node, que modelaindividualmente las máquinas que forman el cluster y albergan los slots queejecutan las tareas de cada job, DFS que representa al filesystem distribuído,y NetworkManager que proporciona la lógica de la red y colecciona estadísti-cas de su utilización. La clase que modela el cluster, además, se utiliza paracalcular la distancia topológica entre nodos.

La modelización de los archivos, utilizados como input por los jobs eje-cutados, está a cargo de la clase File. A partir del tamaño del archivo, deltamaño de bloque del filesystem y del factor de replicación, se generan lossplits correspondientes y se asignan a distintos nodos del cluster de formaaleatoria, de la misma manera en que lo hace Hadoop.

Asimismo, las jerarquía de clases Slot, MapSlot y ReduceSlot implemen-tan la lógica de ejecución de las tareas Map y Reduce en los nodos, mien-tras que coleccionan estadísticas. Las clases SlotManager, MapSlotManagery ReduceSlotManager forman una jerarquía análoga y funcionan como inter-faz entre el nodo y los slots de Map y Reduce alojados en él. Job es la claseencargada de modelar y concentrar la información relativa a los jobs. En ellase definen las propiedades estadísticas de los jobs. Además de gestionar elestado de las tareas Map y Reduce que lo conforman, colecciona estadísticasde ejecución.

Finalmente, la clase abstracta Task controla el estado en el que se en-cuentra cada una de las tareas. Las tareas Map y Reduce se encuentranimplementadas por las clases MapTask y ReduceTask, respectivamente, queextienden de la clase abstracta. Ellas se encargan, a su vez, de implementar lalógica del cambio de estados, cuando en ambos casos es distinta. ReduceTask,adicionalmente, contempla la lógica del transporte de datos intermedio des-de los nodos que ejecutaron las tareas Map hacia el nodo que ejecuta lacorrespondiente tarea Reduce.

La simulación de la ejecución, es decir el programa principal del simula-dor, se encuentra implementado en el script mrschedulersim.py. El mismoinicializa las variables necesarias para llevar adelante el experimento y paracapturar los resultados del mismo. Durante cada ciclo (cuya duración simu-lada es de un segundo), se calcula si corresponde o no el arribo de un nuevojob, se actualiza el estado de todas las tareas Map y Reduce, y se computa suasignación. Las clases Fair, Coupling y Flexible implementan el algoritmode cada uno de los schedulers considerados en la experimentación.

102

La ejecución del programa puede recibir una serie de parámetros paraespecificar las condiciones del experimento: la cantidad de jobs a ejecutar,el número de nodos del cluster, la media y la varianza de la proporción deresultados intermedis generados por las tareas Map a partir de su input,el número máximo de nodos que puede haber en cada rack, el número depuertos que contiene cada switch, y los tamaños posibles para los archivosque serán procesados por los jobs.

5.2.3. Resultados

La figura 5.1 resume los resultados del trabajo. Utilizando el flowtimecomo métrica principal para evaluar la performance de cada scheduler, surgede la experimentación que Flexible Coupling Scheduler obtiene la mejor per-formance, reduciendo el tiempo de procesamiento de los jobs en un 4,78%respecto a Coupling Scheduler, y en un 33,7% respecto a Fair Scheduler.

A continuación se exponen, en mayor detalle, los resultados obtenidosdurante la experimentación para cada un de los esquemas propuestos en laSubsección 5.2.1. En favor de la claridad, se hará referencia a lo largo delpresente capítulo a Flexible Coupling Scheduler simplemente como Flexible,y a los esquemas utilizados para comparar, Fair Scheduler y Coupling Sche-duler, como Fair y Coupling respectivamente.

0

200

400

600

800

724,

34

504,

32

480,

22

Tiempo

deejecuc

ión(s)

FairCouplingFlexible

Figura 5.1: Valor promedio del tiempo de procesamiento de los jobs.

103

Esquema I. Ambiente de producción

En primer lugar, se decidió evaluar la performance de cada uno de losschedulers considerados en un contexto de funcionamiento que resulta fre-cuente en las implementaciones productivas de MapReduce. Esta situaciónconsiste en el arribo secuencial de 100 jobs, cuyos tamaños de input varíanentre 2 y 32 GB, y donde la cantidad de tiempo entre arribos consecutivosse encuentra distribuída exponencialmente, en sintonía con la modelizaciónrealizada por Jian Tan et al. [107]. El cluster se encontraba compuesto por66 nodos, valor típico según publican Abhishek Verma et al. [143].

La ejecución de la simulación para este experimento fue repetida 50 ve-ces con el objetivo de obtener, para cada una de las métricas consideradas,un promedio y la desviación estándar del resultado de cada scheduler. Deesta forma es posible mitigar el factor aleatoreo del experimento y obtenerresultados más precisos.

Específicamente, el intervalo de tiempo Ii (medido en segundos) entre losarribos de dos jobs consecutivos Ji−1 e Ji se encuentra distribuído de formaexponencial con tasa λ = 0,1. El tamaño |Mi| del input del job Ji, medidoen GB, puede tomar alguno de los valores del conjunto CI = 2, 4, 8, 16, 32,con igual probabilidad.

Para comparar la performance bajo este esquema de experimentaciónde los schedulers considerados, en la figura 5.1 se muestran los valores delflowtime obtenidos por cada uno de ellos. Fair obtiene el mayor promediode tiempo ejecución de jobs, con 724,34 unidades de tiempo (que en la si-mulación están modeladas como segundos). Coupling scheduler obtiene unpromedio de ejecución de 504,32 segundos, lo que representa una mejoría del30,38% respecto a Fair, resultado similar al obtenido por sus autores [98].

Puede observarse asimismo que la performance de Flexible supera la de losotros dos schedulers considerados. Con un promedio de ejecución de 480,22segundos, reduce el tiempo de Fair en un 33,7% y el de Coupling en un4,78%, validando el análisis teórico presentado en el Capítulo 4.

El gráfico de la figura 5.2 expone los valores obtenidos del makespan paracada scheduler. En este caso, el tiempo que toma Coupling para procesartodos los jobs es de 3015,9 segundos con una desviación estándar de 158,01segundos. Flexible toma en cambio 2891,1 segundos (con una desviaciónestándar de 167,87 segundos) para completar el conjunto de jobs ejecutados,lo que implica una mejora de 4,14%.

104

0

1000

2000

3000

Tiempo

deejecuc

ión(s)

FairCouplingFlexible

Figura 5.2: Valor promedio y desviación estándar del tiempo total de ejecución.

No obstante Fair obtiene un valor de makespan que supera a ambos, conun tiempo de ejecución total de 2562,7 segundos y una desviación estándar de141,8 segundos. Ello representa una mejora de 11,38% respecto de Flexible,y de 15,03% respecto de Coupling.

Asimismo, durante la experimentación se midió, para cada instante detiempo, la cantidad de conexiones establecidas entre TaskTrackers ejecutandotareas Reduce y TaskTrackers ejecutando tareas Map del mismo job. Cadapar <p, q> de tareas Map p ∈Mi y tareas Reduce q ∈ Ri del job Ji estableceuna conexión para transmitirse por medio de la red del cluster, en caso deser necesario, los datos intermedios producidos en p que le corresponden a q,durante el tiempo que haga falta para transmitirlos.

De esta forma se calculó el promedio de la cantidad de conexiones abiertaspor instante simulado de tiempo, de manera tal de obtener una medida deltráfico de la red. La ocupación o tráfico en la red por parte de cada scheduleren este experimento queda plasmada entonces en gráfico de la figura 5.3.

Puede observarse que Flexible es el scheduler que más ocupación de redacarrea, con 700,44 conexiones por unidad de tiempo con una desviaciónestándar de 52,7. Sólo un 1,17% debajo, según las mediciones obtenidas, Fairhace un uso de la red de 692,22 conexiones por unidad de tiempo y desviaciónestándar de 32,45. Coupling, por su parte, registra una ocupación de red de596,76 conexiones por unidad de tiempo (con una desviación estándar de30,98), es decir un 14,8% por debajo de Flexible y un 13,79% menos que elvalor obtenido por Fair.

105

0

200

400

600

800

Con

exione

spo

rsegu

ndo

FairCouplingFlexible

Figura 5.3: Nivel de tráfico en la red.

Es necesario destacar que estos valores sólo tienen en cuenta el uso dela red por parte de las tareas Reduce. Se ignora aquí el transporte de splitsrealizado por las MapTasks en tanto que se asume el mismo sistema de asig-nación de tareas Map para los tres schedulers. Cualquier virtud o perjuicioque pudiera implicar un cambio de esquema de asignación de tareas Map severía reflejado de la misma forma en la performance de los schedulers, conlo cual en términos comparativos no alteraría los resultados.

La última métrica a considerar es el nivel de localidad de datos. La figura5.4 presenta los resultados obtenidos durante la experimentación, presenta-dos como el valor promedio de distancias topológicas H(ϑi(p), ξi(q)) entre losnodos ϑi(p) y ξi(q), que albergan respectivamente la ejecución de las tareasp ∈Mi y q ∈ Ri del job Ji, para cada job ejecutado.

El promedio de distancias topológicas en el caso de Fair Scheduler, segúnlos resultados obtenidos, es de 3,2340, con una variación estándar menor a0,01 en las mismas unidades. Flexible Coupling Scheduler logra una mejoríade 1,56% respecto a dicho scheduler, obteniendo un promedio de 3,1836 yuna variación estándar también menor a 0,01.

El mejor resultado para esta métrica lo obtiene, sin embargo, CouplingScheduler, que mejora el nivel de Flexible en un 0,15% alcanzando un pro-medio de distancias topológicas de 3,1788 (y cuya variación estándar es asi-mismo menor a 0,01). La diferencia entonces entre Coupling y Fair es por lotanto de 1,71%.

106

3,16

3,18

3,2

3,22

3,24

Distancia

entretareas

FairCouplingFlexible

Figura 5.4: Nivel de localidad de datos.

Finalmente se considera el detalle del nivel de flowtime resultante, paracada scheduler, de acuerdo al tamaño de input de los jobs. En la figura 5.5se puede observar que, para los jobs pequeños (es decir, con un tamaño deinput de entre 2 y 8 GB) los schedulers que disparan de forma gradual lastareas Reduce tienen una performance significativamente superior a Fair, ymuy similar entre sí.

Para los jobs con un input de 2 GB, la diferencia entre la performance deCoupling y Flexible es de 0,12%, con ventaja para este último. En cambio, lamejoría de performance de ambos con respecto Fair es de aproximadamente66% según el resultado de la experimentación.

En el caso de los jobs con input de 4 y 8 GB se da una situación similar:la mejoría de Coupling y Flexible, respecto a Fair, es de aproximadamente65% y 63%, la mejoría de Flexible respecto a Coupling es de 0,28% y 1,5%,respectivamente.

En el caso de los jobs cuyo tamaño de input es 16 GB se tiene que ladiferencia entre Flexible y Coupling se comienza a agrandar, alcanzandoun 4,49% en favor del scheduler presentado como contribución del presentetrabajo. A su vez, el margen de mejora de Flexible respecto a Fair se acortaa un valor de 54,49%.

Para los jobs de mayor tamaño (32 GB de input), en cambio, la mejorperformance la registra Fair, obteniendo una mejora de 34,36% respecto aFlexible, y de 38,69% respecto de Coupling. Asimismo, la performance deFlexible supera a la de Coupling en un 6,75%.

107

2 4 8 16 320

200

400

600

800

1000

1200

1400

1600

Tamaño del input (GB)

Tiempo

deejecución(s)

FairCouplingFlexible

Figura 5.5: Tiempo de procesamiento de los jobs según su tamaño de input.

108

Esquema II. Variación de las condiciones

El segundo esquema de experimentación contempla diversas situaciones,en donde se varían por separado dos parámetros para observar su implicanciaen los resultados: la cantidad N de nodos del cluster y la tasa λ de arribosde jobs al sistema.

Al igual que en la prueba anterior, cada uno de los experimentos queconforman las dos progresiones de este esquema se repitió 50 veces paraobtener el resultado promedio y la desviación estándar para cada scheduler,y mitigar así la incidencia de los factores aleatorios de la simulación.

Durante la primera progresión, se ejecutó el simulador utilizando las mis-mas condiciones iniciales que en la prueba anterior, excepto por la cantidadN de nodos del cluster, con N ∈ 32, 64, 126, 256. Al variar la cantidad denodos que conforman el cluster, se obtienen los siguientes resultados.

En la figura 5.6(a) se grafica la variación del flowtime o promedio detiempo de ejecución de los jobs en función del tamaño del cluster. Puedenotarse que el valor de dicha métrica para todos los casos de N mantiene elmismo orden entre los schedulers, siendo Flexible el de mejor performance,y Fair el de mayor valor.

Asimismo se registra que a mayor valor de N , menor es el tiempo de res-puesta para todos los schedulers, lo que es esperable en el rango consideradodado que se dispone de mayor paralelismo para procesar el trabajo de cadajob. La diferencia de performance entre Flexible y Coupling es similar paratodos los valores de N , no obstante la discrepancia del valor de ambos conrespecto a los de Fair se achica a medida que aumenta la cantidad de nodos.

Los resultados referentes al makespan se exponen en el gráfico que apareceen la figura 5.6(b). En este caso también se mantiene inalterado el orden delos tres schedulers, donde Flexible resulta ser el de mejor performance, yCoupling el que tiene el valor más alto.

Sucede igualmente en este caso que a mayor N , menor es el valor demakespan obtenido para cada uno de los schedulers. En contraste con elresultado para la métrica anterior, la diferencia de valores obtenidos porCoupling y Flexible (con una leve ventaja del esquema propuesto como con-tribución del presente trabajo) se reduce conforme aumenta la cantidad denodos del cluster, mientras que la discrepancia de sus valores respecto a losde Fair Scheduler se amplía.

109

32 64 128 2560

500

1000

1500

2000

Cantidad N de nodos del cluster

Tiempo

deejecuc

ión(s) Fair

CouplingFlexible

(a) Flowtime.

32 64 128 2560

2000

4000

Cantidad N de nodos del cluster

Tiempo

deejecución(s) Fair

CouplingFlexible

(b) Makespan.

Figura 5.6: Impacto de la cantidad de nodos del cluster sobre la performance, en términosde (a) flowtime y (b) makespan.

110

El gráfico la figura 5.7(a) expone los resultados referentes a la ocupaciónde la red en función de la variación de la cantidad de nodos del cluster,calculado de la misma manera que en el experimento de la figura 5.3. Elorden de los schedulers se mantiene invariante también aquí, siendo Couplingel de menor uso de red, y Fair el de mayor ocupación.

Conforme crece la cantidad de nodos N del cluster, mayor es el nivelde uso de la red para los tres schedulers. Los valores de Flexible tienden aacercarse a aquellos de Coupling, a medida que aumenta N (con excepcióndel caso donde N es 64, que de todas formas registra una alteración sutil),mientras que ambos se distancian de los valores de Fair Scheduler.

En la figura 5.7(b) se plasman los resultados obtenidos para el nivel delocalidad de datos para las tareas Reduce. Al igual que para las demás mé-tricas, se mantiene sin variaciones el orden de los schedulers: el mejor nivelde localidad de datos lo obtiene Coupling, pero apenas por debajo le sigeFlexible, quedando Fair con el nivel más bajo de localidad de datos.

El nivel de localidad de datos para las tareas Reduce disminuye (es decir,aumenta la distancia topológica promedio entre tareas Map y Reduce delmismo job) en tanto que aumenta el tamaño del cluster. Los valores paraFlexible y Coupling son muy similares, con una ligera ventaja para esteúltimo, mientras que la diferencia de ambos con respecto a los valores deFair aumenta a mayor N .

La última progresión consta de experimentos en los cuales, a partir de lasmismas condiciones que en el esquema inicial, se varía el valor de la tasa λde arribo de los jobs sobre la performance del sistema, considerando que eltiempo transcurrido entre la llegada de dos jobs consecutivos al sistema sedistribuye de manera exponencial. Para ello se toman valores de la tasa dearribos tales que λ ∈ [0,02; 0,09].

La tasa λ afecta a la cantidad de jobs que se encontrarán en servicio(ejecutándose o en espera) en un momento dado, en tanto que si el tiempopromedio entre la llegada de dos jobs consecutivos es mayor, el sistema deberádistribuír sus recursos entre un número menor de jobs durante más tiempo.

El gráfico de la figura 5.8(a) muestra el impacto de la variación de latasa de arribos λ sobre la peformance del sistema, medida en términos deltiempo promedio requerido para ejecutar un job. Flexible resulta ser el demejor performance en todos los casos de λ, en tanto que Fair es siempre elde mayor tiempo de respuesta.

111

32 64 128 256

400

600

800

1000

1200

Cantidad N de nodos del cluster

Nivel

deusode

lared

FairCouplingFlexible

(a) Ocupación de la red.

32 64 128 256

2,5

3

3,5

4

4,5

Cantidad N de nodos del cluster

Distanc

iatopo

lógica

prom

edio

FairCouplingFlexible

(b) Nivel de localidad de datos.

Figura 5.7: Impacto de la cantidad de nodos del cluster sobre (a) la ocupación de red y(b) el nivel de localidad de datos.

112

Además de mantenerse invariante el orden de los schedulers en cuanto ala performance obtenida para cada valor de la tasa de arribos, cabe destacarque la diferencia del valor entre cada uno de también permanece estable. Lamejor performance, para todos los schedulers, se verifica en el valor más bajode λ, mientras que el mayor tiempo de respuesta se obtiene en algún puntodel intervalo (0,06; 0,08).

La variación del makespan, para cada uno de los schedulers, en funcióndel valor de λ se muestra en el gráfico de la figura 5.8(b). En este caso(al igual que en el primer experimento) la mejor performance la obtieneFair Scheduler, mientras que el mayor tiempo de respuesta le corresponde aCoupling. La diferencia entre la performance de Coupling y de Flexible esademás mayor a la que registrada en el caso del flowtime, con ventaja parael scheduler propuesto como contribución de la presente tesis.

Respecto a los demás aspectos a considerar, la variación del makespanresponde a las mismas características que para el gráfico anterior: el ordende los schedulers para esta métrica no varía, la diferencia del valor obtenidopor cada uno permance estable para todos los casos de λ, y los máximos ymínimos encontrados se producen en las mismas regiones del gráfico.

La figura 5.9(a) expone el impacto de la tasa de arribos λ sobre el nivelde ocupación de la red. El uso de la red por parte de Coupling Scheduleres claramente inferior al de los otros dos schedulers, en tanto que Fair tieneun nivel de ocupación levemente menor que Flexible, con excepción del casoλ = 0,02 (posiblemente debido a factores aleatorios en la simulación).

La distancia entre Coupling respecto a los otros dos schedulers, Flexibley Fair, es significativamente mayor que la distancia entre estos dos últimos.Asimismo se observa que no hay un punto claro de máximo ni de mínimorendimiento para ninguno de los tres schedulers, dado que la variación entresucesivas ejecuciones del experimento es mayor que la diferencia entre elpromedio de cada scheduler para distintos valores de λ.

El impacto de λ sobre el nivel de localidad de datos alcanzado por cadascheduler puede ser observado en el gráfico de la figura 5.9(b). El orden delos schedulers se mantiene para todos los valores de la tasa de arribo: el me-jor nivel de localidad de datos lo obtiene Coupling, seguido muy cerca porFlexible. Por otro lado, Fair Scheduler tiene un promedio de distancia topo-lógica entre tareas Map y Reduce del mismo job significativamente mayorque ambos.

113

0,02 0,03 0,04 0,05 0,06 0,07 0,08 0,09 0,10

400

500

600

700

800

900

Tasa de arribos (λ)

Tiempo

deejecuc

ión(s)

FairCouplingFlexible

(a) Flowtime.

0,02 0,03 0,04 0,05 0,06 0,07 0,08 0,09 0,102200

2400

2600

2800

3000

3200

3400

Tasa de arribos (λ)

Tiempo

deejecuc

ión(s)

FairCouplingFlexible

(b) Makespan.

Figura 5.8: Impacto de la tasa de arribos de jobs sobre la performance, en términos de(a) flowtime y (b) makespan.

114

5.3. Análisis de los resultados

En la presente sección se resaltan algunas reflexiones sobre resultadosobtenidos durante los distintos esquemas de experimentación detallados, yse contrastan dichos resultados con lo establecido en el capítulo Capítulo 4en relación a la performance del scheduler propuesto.

Tomando en cuenta Coupling Scheduler y Flexible Coupling Scheduler,se tiene que Flexible alcanza una mejor performance tanto en términos deflowtime, en promedio y para cada uno de los casos de tamaño de carga detrabajo de los jobs, como en térmnos de makespan. Esta mejoría se obtiene aexpensas del tráfico en la red en tanto que que, si se considera cualquier lapsode tiempo durante el funcionamiento del sistema, Flexible siempre asigna unacantidad igual o superior de tareas Reduce que Coupling.

Siempre y cuando no termine colapsando la red del cluster, Flexible estaráhaciendo un mejor aprovechamiento de los recursos del cluster. El esquemade lanzamiento gradual de tareas Reduce, diseñado por los creadores deCoupling, establece que el mecanismo para evitar el colapso de la red esel ritmo de incorporación de tareas a la cola Q de espera, utilizando unafunción ψ para comparar el progreso de las fases Map y Reduce de cadajob. Los ciclos de heartbeats que deja pasar Coupling sin realizar ningunaasignación esperando a encontrar el mejor posicionamiento para la tareacandidata son retrasos adicionales a dicho esquema.

Las asignaciones adicionales que realiza Flexible implican asimismo unaoptimización, debido al mejor aprovechamiento de los recursos del clusteren detrimento del nivel de localidad de datos alcanzado por el scheduler(Coupling, para garantizar el nivel de localidad de datos, puede llegar aesperar hasta 4N hearbeats). No obstante, este detrimento demuestra sermuy pequeño según los resultados de la experimentación. Obsérvese que ladisminución del nivel de localidad de datos en Flexible respecto a Couplingse minimiza al utilizar el estimador Ω, dado que, para asignar una tareaReduce adicional a las contempladas por Coupling Scheduler, selecciona eljob que menos perjudique al nivel de localidad de dtos.

Si se compara en cambio el scheduler propuesto con respecto a Fair Sche-duler (el más utilizado en la actualidad en implementaciones productivas deMapreduce), se tiene que Flexible registra una mejor performance en tér-minos de flowtime, en promedio y particularmente para los jobs de tamañopequeño y mediano.

115

0,02 0,03 0,04 0,05 0,06 0,07 0,08 0,09 0,10550

600

650

700

750

Tasa de arribos (λ)

Nivel

deusode

lared

FairCouplingFlexible

(a) Ocupación de la red.

0,02 0,03 0,04 0,05 0,06 0,07 0,08 0,09 0,10

3,18

3,20

3,22

3,24

Tasa de arribos (λ)

Distancia

topo

lógica

FairCouplingFlexible

(b) Nivel de localidad de datos.

Figura 5.9: Impacto de la tasa de arribos de jobs sobre (a) la ocupación de red y (b) elnivel de localidad de datos.

116

Por otro lado, la performance de Fair es mejor en términos de flowtimepara job cuyo input es grande, y en términos de makespan. La manera es-calonada de lanzar las tareas Reduce de Flexible logra mejorar el promediode tiempo de finalización de los jobs más pequeños [98], en desmedro de laperformance de los jobs más grandes. El hecho de tener un mejor flowtimepara jobs grandes tiene un mayor impacto sobre el makespan, que representael tiempo desde el comienzo del primer job hasta la finalización de aquel quetermine último, con lo cual se explica la mejoría de performance según estamétrica respecto a Flexible.

No obstante, obtener mejor performance para los jobs de tamaño pequeñoy mediano tiene un impacto mayor sobre el flowtime general, dado que la dis-tribución de probabilidad del tamaño de los jobs en entornos de produccióntiende a ser heavy-tailed [100, 109]. Cabe notar que esta mejoría de Flexiblerespecto a Fair es compartida también por Coupling (se debe al lanzamientogradual de tareas Reduce) y se analiza en mayor detalle en la publicación deJian Tan et al. [97, 98].

117

6

Conclusiones y trabajo futuro

Durante la presente disertación se realizó un detallado análisis del funcio-namiento de MapReduce y del estado del arte en materia de administraciónde sus recursos, y se desarrolló un modelo formal para estudiar el problemade optimización del componente de scheduling. El scheduler presentado co-mo contribución, que maximiza el nivel de localidad de datos de las tareasReduce y evita subutilizar recursos del cluster, se basa en las optimizaciónesmás recientes.

Utilizando el esquema de lanzamiento gradual de tareas Reduce y el mé-todo de asignación de Coupling Scheduler, el algoritmo presentado agregaademás una asignación extra cada N heartbeats recibidos en caso de nopoder ubicar una tarea del job con mayor valor de mismatch (siendo N lacantidad de nodos de nodos del cluster).

Para comparar los jobs con tareas Reduce pendientes a ser ejecutadas,y seleccionar uno de ellos para que reciba aquella asignación adicional, seutiliza un estimador que evalúa el potencial ahorro de transporte de datosa través de la red, en base al nivel de localidad de datos de las tareas queserían ejecutadas.

Mediante simulación, se obtuvieron además resultados experimentales pa-ra evaluar la contribución y validar lo establecido durante el análisis. En estecapítulo se exponen las conclusiones de la tesis, que figuran en la Sección 6.1,y finalmente se describen en la Sección 6.2 las perspectivas de la contribucióny posibles direcciones para continuar con su investigación.

118

6.1. Conclusiones

Los resultados obtenidos confirman la validez de lo que fue planteado enel Capítulo 4. El componente de scheduling de MapReduce propuesto comocontribución de la tesis, denominado Flexible Coupling Scheduler, logramejorar el flowtime o tiempo promedio de procesamiento de los jobs ejecu-tados respecto a los algoritmos existentes, en tanto que mantiene un buennivel de localidad de datos y al mismo tiempo evita incurrir en subutilizaciónde recursos.

Asimismo, se refuerzan también las conclusiones de la investigación reali-zada por Jian Tan et al. [98], en donde se afirma que el lanzamiento gradualde tareas Reduce conforme progresa la fase Map, de forma tal de optimi-zar en forma conjunta la localización de ambos tipos de tareas, mitiga elproblema de inanición de los jobs y mejora el nivel de localidad de datos.

Tanto el scheduler propuesto, como el exhaustivo análisis del estado delarte y el modelo formal desarrollado, podrían resultar de utilidad como puntode referencia para futuras investigaciones sobre la administración eficiente derecursos en MapReduce y en otros frameworks similares de procesamientomasivo de datos.

6.2. Futuras líneas de investigación

En la siguiente lista se detallan algunas líneas de investigación y desarrollopropuestas para continuar con el trabajo realizado.

Implementación en Hadoop. Si bien la implementación de FlexibleCoupling Scheduler como contribución a Hadoop se encuentra fuera delalcance de una tesis de grado (dado que el código fuente de CouplingScheduler [98] no es público), representa una de las principales direc-ciones para continuar con el trabajo aquí expuesto.

Evaluación de la implementación. Una vez implementado el al-goritmo como plug-in de Hadoop, se abre la posibilidad de evaluarlodirectamente a través de experimentaciones sobre la propia plataformay comparar su performance respecto a la de otros schedulers medianteel uso benchmarks, y para encontrar además nuevos espacios de mejoray continuar perfeccionándolo.

119

Perfeccionamiento del simulador. El simulador mrschedulersim,desarrollado con el fin de evaluar la contribución de la tesis, podría me-jorarse implementando un sistema más complejo de gestión de trans-porte de datos en el cluster. Específicamente, la clase NetworkManagerdel simulador podría ser reimplementada para alterar el ancho de ban-da brindado a cada una de las tareas que utilizan la red, de acuerdo ala cantidad de carga que hay en la misma.

Extensión a Spark. Adaptar el algoritmo de scheduling presentadopara Spark representa otra perspectiva interesante de investigación, entanto que dicha plataforma obtiene excelente performance al ejecutarjobs iterativos y se encuentra en constante desarrollo.

120

Apéndice A

Nomenclatura

En el presente anexo se mapean los conceptos desarrollados en el modelodel problema y en la solución propuesta, presentados en el Capítulo 4, conlos símbolos que se utilizan para representarlos.

Símbolo Concepto representado

G Topología del cluster, compuesta por los conjuntos V y E .

V Conjunto de nodos del cluster.

u, v Elementos utilizados para hacer referencia a nodos genéricos delcluster.

E Conjunto de conexiónes entre nodos del cluster.

H(u,v) Función de distancia topológica entre dos nodos u,v ∈ V.

J Conjunto de jobs ejecutados, indexados según su orden de arribo.

iÍndice utilizado para hacer referencia a un dado job del conjuntoJ .

Mi Conjunto de tareas Map del job Ji.

Mi Cantidad de tareas Map del job Ji.

121

Símbolo Concepto representado

p Elemento utilizado para hacer referencia a una tarea Map genérica.

Ri Conjunto de tareas Reduce del job Ji.

Ri Cantidad de tareas Reduce del job Ji.

q, rElementos utilizados para hacer referencia a tareas Reduce gené-rica.

| · |Operador para representar la cantidad de trabajo, medida a travésdel volúmen total del input, de una tarea Map o Reduce, o de unconjunto de tareas.

Xi Volúmen total de datos intermedios del job Ji.

Xi(u)Volúmen de datos intermedios del job Ji almacenados en el nodou ∈ V.

XqVolúmen total de datos intermedios destinados a la tarea Reduceq ∈ Ri.

Xq(u)Volúmen de datos intermedios destinados a la tarea Reduce q ∈ Riy almacenados en el nodo u ∈ V.

Xpq

Volúmen total de datos intermedios producidos por la tarea Mapp ∈Mi y destinados a la tarea Reduce q ∈ Ri.

Xpq (u)

Volúmen de datos intermedios producidos por la tarea Map p ∈Mi, destinados a la tarea Reduce q ∈ Ri y almacenados en elnodo u ∈ V.

XiVolúmen parcial de datos intermedios del job Ji, que fueron gene-rados antes de un dado momento t.

Xi(u)Volúmen parcial de datos intermedios del job Ji almacenados en elnodo u ∈ V, que fueron generados antes de un dado momento t.

XqVolúmen parcial de datos intermedios destinados a la tarea Reduceq ∈ Ri, que fueron generados antes de un dado momento t.

122

Símbolo Concepto representado

Xq(u)Volúmen parcial de datos intermedios destinados a la tarea Reduceq ∈ Ri y almacenados en el nodo u ∈ V, que fueron generados antesde un dado momento t.

Xpq

Volúmen parcial total de datos intermedios producidos por la tareaMap p ∈ Mi y destinados a la tarea Reduce q ∈ Ri, que fuerongenerados antes de un dado momento t.

Xpq(u)

Volúmen parcial de datos intermedios producidos por la tarea Mapp ∈Mi, destinados a la tarea Reduce q ∈ Ri y almacenados en elnodo u ∈ V, que fueron generados antes de un dado momento t.

ϑiFunción que mapea cada tarea Map del job Ji al nodo u ∈ V dondees asignada.

ξiFunción que mapea cada tarea Reduce del job Ji al nodo u ∈ Vdonde es asignada.

C(q)Costo total para transferir los datos intermedios de la tarea Reduceq ∈ Ri al nodo ξi(q) donde es asignada.

A Conjunto de archivos almacenados en el file system distribuído,que sirven de input para los jobs de J .

sÍndice utilizado para hacer referencia a un dado archivo del con-junto A.

Bs Conjunto de bloques que confiorman el archivo As.

ls Cantidad de bloques que conforman el archivo As.

ρ Factor de replicación por defecto del cluster.

ρs Factor de replicación específico para los bloques del archivo As.

Sm Conjunto de todos los MapSlots del cluster.

Sm(u) Conjunto de los MapSlots del nodo u ∈ V.

Sr Conjunto de todos los ReduceSlots del cluster.

Sr(u) Conjunto de los ReduceSlots del nodo u ∈ V.

123

Símbolo Concepto representado

Dt(·)Operador que, aplicado sobre un conjunto de slots, obtiene el sub-conjunto formado por todos aquellos en el instante t se encuentrandisponibles.

IiIntervalo de tiempo entre los arribos de dos jobs consecutivos Ji−1

y Ji.

KCantidad máxima de jobs que puede haber simultáneamente enejecución.

ψProporción de tareas Reduce incorporadas a la cola correspondien-te, en función del progreso de la fase Map.

xi Progreso de la fase Map en un dado instante t.

yi Progreso de la fase Reduce en un dado instante t.

Ω(v, Ji)

Estimación del costo total de transferencia de resultados interme-dios para una tarea Reduce promedio del job Ji ejecutada en elnodo u ∈ V, calculada con los datos disponibles en un dado ins-tante t.

κ(u)Estimación del tamaño total de datos intermedios del job Ji alma-cenados en el nodo u ∈ V, calculado con los datos disponibles enun dado instante t.

γCuadrado del tamaño total estimado de datos intermedios del jobJi.

Q Cola de tareas Reduce pendientes a ser ejecutadas.

Cuadro A.1: Nomenclatura utilizada para la contribución del Capítulo 4

124

Apéndice B

Código fuente

A continuación figura el código fuente del simulador mrschedulersim,escrito en Python. El mismo fue desarrollado con el objetivo de evaluar laperformance del scheduler presentado como contribución de la tesis.

Script B.1: mrschedulersim.py

1 #!/usr/bin/python2

3 """4 Copyright 2015 Federico Retyk5

6 Licensed under the Apache License, Version 2.0 (the "License");7 you may not use this file except in compliance with the License.8 You may obtain a copy of the License at9

10 http://www.apache.org/licenses/LICENSE-2.011

12 Unless required by applicable law or agreed to in writing, software13 distributed under the License is distributed on an "AS IS" BASIS,14 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.15 See the License for the specific language governing permissions and16 limitations under the License.17 """18

19 import math20 import random21 import statistics22 import sys23 from collections import deque

125

24 from argparser import getArgParser25 from cluster import Cluster26 from datetime import datetime27 from dfsfile import File28 from job import Job29 from mrlogging import log30 from procedures import generateJob31 from procedures import stdev32 from schedman import SchedulerManager33 from task import Task34

35 argparser = getArgParser()36 args = argparser.parse_args()37

38 # time counter for the simulation (in seconds)39 currentTime = 040

41 # initialize cluster and distributed file system properties42 cluster = Cluster(args.nodes, args.nodesxrack, args.switchports)43

44 # initialize schedulers45 jobTracker = SchedulerManager(cluster, args.psi)46

47 # job properties48 numberOfJobs = args.jobs49 Job.meanRatio = args.redinput[0]50 Job.stdevRatio = args.redinput[1]51

52 # initialize other variables53 simCompleted = False54 nextJob = 055 executedJobs = 056 jobs = []57 files = []58 pendingMapQueue = deque()59 pendingReduceQueue = 60

61 # generate files for the experiment62 fileSizes = args.fsizes63 for s in fileSizes:64 files.append(File(s, cluster))65

66 # initiate auxiliary variables for performance observation67 jobCompletionTimes = 68 jobCompletionTimesPerSize = 69 for sched in SchedulerManager.schedulerList:70 pendingReduceQueue[sched] = deque()71 jobCompletionTimes[sched] = []72 jobCompletionTimesPerSize[sched] =

126

73 for size in fileSizes:74 jobCompletionTimesPerSize[sched][size] = []75 jobCompletionTimes[’map’] = []76

77 # start simulation78 startTime = datetime.now()79

80 while not simCompleted:81

82 cluster.network.update()83

84 if currentTime == nextJob:85 executedJobs += 186 generateJob(jobs, files, currentTime)87 for mapTask in jobs[len(jobs)-1].getMapTasks():88 pendingMapQueue.append(mapTask)89 mapTask.queue(currentTime)90 if (executedJobs == numberOfJobs):91 nextJob = -192 else:93 nextJob += 1 + math.ceil(random.expovariate(Job.submRate))94

95 # data-local MapTask assignment96 mapQueueLength = len(pendingMapQueue)97 for i in range(mapQueueLength):98 mapTask = pendingMapQueue.popleft()99 assigned = False

100 for localNode in mapTask.localNodes:101 currentNode = cluster.nodes[localNode]102 if currentNode.hasAvailableMapSlots():103 currentNode.assignMapTask(mapTask, currentTime)104 assigned = True105 break106 if not assigned:107 pendingMapQueue.append(mapTask)108

109 for node in cluster.shuffledNodes:110 # update status for slots111 node.updateSlots(currentTime)112 # assign non-data-local MapTask from queue113 while (node.hasAvailableMapSlots() and len(pendingMapQueue)):114 mapTask = pendingMapQueue.popleft()115 node.assignMapTask(mapTask, currentTime)116

117 # assign ReduceTask from queue118 jobTracker.assignReduceTasks(pendingReduceQueue, currentTime)119

120 # update shuffle status for running ReduceTask121 for sched in SchedulerManager.schedulerList:

127

122 for job in jobs:123 for reduceTask in job.getReduceTasks(sched, Task.State.Running):124 reduceTask.performShuffle()125

126 # queue ReduceTasks127 jobTracker.queueReduceTasks(jobs, pendingReduceQueue, currentTime)128

129 completedJobs = [job for job in jobs if job.isCompleted(currentTime)]130 for job in completedJobs:131 jobCompletionTimes[’map’].append(job.durations[’map’])132 for sched in SchedulerManager.schedulerList:133 jobCompletionTimes[sched].append(job.durations[sched])134 compTimes = jobCompletionTimesPerSize[sched][job.inputFile.size]135 compTimes.append(job.durations[sched])136 jobs[:] = [job for job in jobs if not job.isCompleted(currentTime)]137 if ((nextJob == -1) and len(jobs) == 0):138 simCompleted = True139 currentTime += 1140 # shuffle cluster nodes list to simulate heartbeat randomness141 cluster.shuffleNodes()142

143

144 endTime = datetime.now()145 duration = endTime - startTime146

147 log(’main’, ’Simulated duration ’ + currentTime + ’ seconds’)148 log(’main’, ’Real duration ’ + str(duration))149

150 meanJCT = statistics.mean(jobCompletionTimes[’map’])151 stdevJCT = stdev(jobCompletionTimes[’map’])152

153 log(’main’, ’Map phase completion time mean of ’ + str(meanJCT))154 log(’main’, ’Map phase completion time stdev of ’ + str(stdevJCT))155

156 for sched in SchedulerManager.schedulerList:157 meanJCT = statistics.mean(jobCompletionTimes[sched])158 stdevJCT = stdev(jobCompletionTimes[sched])159 log(’main’, ’Statistical information for scheduler ’ + sched)160 log(’main’, ’Job completion time mean of ’ + str(meanJCT))161 log(’main’, ’Job completion time stdev of’ + str(stdevJCT))162 for size in fileSizes:163 compTimes = jobCompletionTimesPerSize[sched][size]164 meanJCTPerSize = statistics.mean(compTimes)165 stdevJCTPerSize = stdev(compTimes)166 log(’main’, ’For jobs with size ’ + str(size))167 log(’main’, ’Job completion time mean of ’ + str(meanJCT))168 log(’main’, ’Job completion time stdev of’ + str(stdevJCT))169 log(’main’, ’’)

128

Script B.2: cluster.py

1 import random2 import statistics3 from dfs import DFS4 from networkmanager import NetworkManager5 from node import Node6

7 class Cluster:8

9 # slots per node10 mSlotsPerNode = 511 rSlotsPerNode = 212

13 def __init__(self, numberOfNodes, nodesPerRack, portsPerSwitch):14 self.generateNodes(numberOfNodes)15 self.nodesPerRack = nodesPerRack16 self.portsPerSwitch = portsPerSwitch17 self.dfs = DFS()18 self.network = NetworkManager()19

20 def rack(self, u):21 # get rack index for a given node with index u22 rackId = u / self.nodesPerRack23 return rackId24

25 def hopDistance(self, u, v):26 # calculates H(u,v) for two node indexes u and v27 if u == v:28 return 029 if self.rack(u) == self.rack(v):30 return 131 h = 232 searchRange = self.nodesPerRack33 while searchRange < len(self.nodes):34 searchRange = searchRange * self.portsPerSwitch35 uRange = u / searchRange36 vRange = v / searchRange37 if (uRange == vRange):38 break39 h += 140 return h41

42 def numberOfNodes(self):43 return len(self.nodes)44

45 def shuffleNodes(self):46 random.shuffle(self.shuffledNodes)47

129

48 def generateNodes(self, numberOfNodes):49 self.nodes = []50 self.shuffledNodes = []51 for i in range(numberOfNodes):52 node = Node(i, Cluster.mSlotsPerNode, Cluster.rSlotsPerNode)53 self.nodes.append(node)54 self.shuffledNodes.append(node)

Script B.3: node.py

1 from slotmanager import MapSlotManager2 from slotmanager import ReduceSlotManager3 from schedman import SchedulerManager4

5 class Node:6

7 def __init__(self, index, mSlots, rSlots):8 self.index = index9 self.mapSlots = MapSlotManager(mSlots, self)

10 self.reduceSlots = 11 for sched in SchedulerManager.schedulerList:12 self.reduceSlots[sched] = ReduceSlotManager(rSlots, self)13

14 def hasAvailableMapSlots(self):15 return self.mapSlots.hasAvailableSlots()16

17 def assignMapTask(self, task, currentTime):18 return self.mapSlots.assignTask(task, currentTime)19

20 def hasAvailableReduceSlots(self, sched):21 return self.reduceSlots[sched].hasAvailableSlots()22

23 def assignReduceTask(self, sched, task, currentTime):24 return self.reduceSlots[sched].assignTask(task, currentTime)25

26 def updateSlots(self, currentTime):27 self.mapSlots.updateStatus(currentTime)28 for sched in SchedulerManager.schedulerList:29 self.reduceSlots[sched].updateStatus(currentTime)

Script B.4: networkmanager.py

1 import sys2 from schedman import SchedulerManager3

4

5 class NetworkManager:

130

6

7 # cluster’s bandwidth in MB8 bandwidth = 100.09 initialUsage = sys.maxint

10

11 def __init__(self):12 self.currentUsage = 13 self.previousUsage = 14 self.statsBand = 15 self.statsUse = 16 for sched in SchedulerManager.schedulerList:17 self.statsBand[sched] = []18 self.statsUse[sched] = []19 self.currentUsage[sched] = 020 self.previousUsage[sched] = NetworkManager.initialUsage21

22 def requestBandwidth(self, sched):23 self.currentUsage[sched] += 124 return self.calculateBandwidth(sched)25

26 def calculateBandwidth(self, sched):27 usage = self.previousUsage[sched]28 return NetworkManager.bandwidth29

30 def update(self):31 for sched in SchedulerManager.schedulerList:32 if (self.currentUsage[sched] != 0):33 self.previousUsage[sched] = self.currentUsage[sched]34 self.currentUsage[sched] = 035 if self.previousUsage[sched] != NetworkManager.initialUsage:36 bandw = self.calculateBandwidth(sched)37 self.statsBand[sched].append(bandw)38 self.statsUse[sched].append(self.previousUsage[sched])

Script B.5: dfsfile.py

1 import random2 from dfs import DFS3

4 class File:5

6 meanSize = 107

8 def __init__(self, size, cluster):9 self.size = size

10 self.cluster = cluster11 self.blocks = File.generateSplits(size, cluster)12

131

13 def numberOfBlocks(self):14 return len(self.blocks)15

16 @staticmethod17 def generateSplits(size, cluster):18 # size is assummed to be in gigabytes19 numberOfBlocks = (size * 1024) / DFS.blockSize20 blocks = []21 nodes = cluster.numberOfNodes()22 factor = cluster.dfs.replicationFactor23 for i in range(numberOfBlocks):24 blocks.append(random.sample(range(nodes), factor))25 return blocks

Script B.6: dfs.py

1 class DFS:2

3 blockSize = 644

5 def __init__(self):6 self.replicationFactor = 3

Script B.7: task.py

1 import sys2 import math3 from dfs import DFS4

5

6 class Task:7

8 class State:9 Pending, Queued, Running, Completed = range(4)

10

11 def __init__(self, index, job, duration):12 self.index = index13 self.job = job14 self.cluster = self.job.cluster15 self.node = None16 self.state = Task.State.Pending17 self.duration = duration18

19 def queue(self, currentTime):20 self.state = Task.State.Queued21 self.queueTime = currentTime22

132

23 def assign(self, node, currentTime):24 self.node = node25 self.state = Task.State.Running26 self.assignTime = currentTime27

28 def complete(self, currentTime):29 self.state = Task.State.Completed30 self.completeTime = currentTime31

32 def calculateStatistics(self):33 if not (self.state == Task.State.Completed):34 return None35 result = []36 result.append(self.assignTime - self.queueTime)37 result.append(self.completeTime - self.assignTime)38 return result39

40 def getNode(self):41 return self.node42

43 def getId(self):44 return [self.job.index, self.index]45

46

47 class MapTask(Task):48

49 def __init__(self, index, job, duration, localNodes):50 Task.__init__(self, index, job, duration)51 self.localNodes = localNodes52

53 def assign(self, node, currentTime):54 Task.assign(self, node, currentTime)55 if not node.index in self.localNodes:56 minHopDist = sys.maxint57 minNode = None58 for localNode in self.localNodes:59 dist = self.cluster.hopDistance(node.index, localNode)60 if (dist < minHopDist):61 minHopDist = dist62 minNode = localNode63 bw = self.cluster.network.bandwidth64 blockSize = self.imputSize()65 propDelay = minHopDist * (blockSize / bw)66 self.duration += propDelay67

68 def complete(self, currentTime):69 Task.complete(self, currentTime)70

71 def imputSize(self):

133

72 return DFS.blockSize73

74

75 class ReduceTask(Task):76

77 def __init__(self, index, job, duration):78 Task.__init__(self, index, job, duration)79 self.shuffledPerc = []80 self.scheduler = None81 for i in range(job.numberMapTasks):82 self.shuffledPerc.append(0.0)83

84 def performShuffle(self):85 if not (self.state == Task.State.Running):86 return None87 for mapTask in self.job.getMapTasks(Task.State.Completed):88 # update shuffled Mapper result percentages89 if (self.shuffledPerc[mapTask.index] < 1):90 u = mapTask.node.index91 v = self.node.index92 hopDist = self.cluster.hopDistance(u, v)93 progPerSecond = 1.094 if hopDist != 0:95 network = self.cluster.network96 bw = network.requestBandwidth(self.scheduler)97 size = mapTask.imputSize()98 imputRatio = self.job.reduceInputRatio99 resultsSize = (size * imputRatio)

100 redInput = resultsSize / self.job.numberReduceTasks101 propDelay = math.ceil(hopDist * (redInput / bw))102 progPerSecond = 1.0103 if propDelay > 1:104 progPerSecond = 1.0 / propDelay105 self.shuffledPerc[mapTask.index] += progPerSecond106

107 def queue(self, scheduler, currentTime):108 Task.queue(self, currentTime)109 self.scheduler = scheduler110

111 def imputSize(self):112 ratio = self.job.reduceInputRatio113 inputSize = self.job.inputFile.size * 1024114 totalReduceInput = ratio * inputSize115 reducers = self.job.numberReduceTasks116 return totalReduceInput / reducers117

118 def shuffleComplete(self):119 for info in self.shuffledPerc:120 if info < 1.0:

134

121 return False122 return True

Script B.8: schedmanager.py

1 from fair import Fair2 from coupling import Coupling3 from flexible import Flexible4

5 class SchedulerManager:6

7 schedulerList = [8 Fair.Key,9 Coupling.Key,

10 Flexible.Key,11 ]12

13 def __init__(self, cluster, psi):14 self.schedulers = 15 Fair.Key: Fair(cluster),16 Coupling.Key: Coupling(cluster, psi),17 Flexible.Key: Flexible(cluster, psi)18 19

20 def get(sched):21 return self.schedulers[sched]22

23 def assignReduceTasks(self, pendingReduceQueue, currentTime):24 for key, scheduler in self.schedulers.iteritems():25 scheduler.assignReduceTasks(pendingReduceQueue, currentTime)26

27 def queueReduceTasks(self, jobs, pendReduceQueue, currentTime):28 for key, scheduler in self.schedulers.iteritems():29 scheduler.queueReduceTasks(jobs, pendReduceQueue, currentTime)

Script B.9: job.py

1 import math2 import random3 import statistics4 from dfsfile import File5 from task import Task6 from task import MapTask7 from task import ReduceTask8 from schedman import SchedulerManager9 from mrlogging import log

10

135

11

12 class Job:13

14 _count = 015

16 # job statistical properties17 mapMeanTime = 2018 mapStdevTime = 519 reduceMeanTime = 6020 reduceStdevTime = 521 meanReducers = 2022 stdevReducers = 1423 meanRatio = 1.024 stdevRatio = 0.525 submRate = 0.126

27 def __init__(self, inputFile, currentTime):28 self.index = Job._count29 self.inputFile = inputFile30 self.cluster = inputFile.cluster31 self.generateMapTasks()32 self.generateReduceTasks()33 self.initializeCompletionTimes()34 self.arrivalTime = currentTime35 self.durations = 36 Job._count += 137

38 def generateMapTasks(self):39 self.numberMapTasks = self.inputFile.numberOfBlocks()40 self.mapTasks = []41 for i in range(self.numberMapTasks):42 duration = Job.randomNormal(43 Job.mapMeanTime,44 Job.mapStdevTime45 )46 localNodes = self.inputFile.blocks[i]47 self.mapTasks.append(MapTask(i, self, duration, localNodes))48

49 def generateReduceTasks(self):50 self.numberReduceTasks = Job.randomNormal(51 self.numberMapTasks / 4,52 Job.stdevReducers53 )54 self.reduceInputRatio = Job.generateReduceInputRatio()55 self.reduceTasks = 56 for sched in SchedulerManager.schedulerList:57 self.reduceTasks[sched] = []58 for i in range(self.numberReduceTasks):59 d = Job.randomNormal(

136

60 Job.reduceMeanTime,61 Job.reduceStdevTime62 )63 for sched in SchedulerManager.schedulerList:64 self.reduceTasks[sched].append(ReduceTask(i, self, d))65

66 def initializeCompletionTimes(self):67 self.completionTimes = 68 for sched in SchedulerManager.schedulerList:69 self.completionTimes[sched] = -170 self.completionTimes[’map’] = -171

72 def getMapTasks(self, state = None):73 result = []74 if state is None:75 for m in self.mapTasks:76 result.append(m)77 return result78 for m in self.mapTasks:79 if m.state == state:80 result.append(m)81 return result82

83 def getReduceTasks(self, sched, state = None):84 result = []85 if state is None:86 for r in self.reduceTasks[sched]:87 result.append(r)88 return result89 for r in self.reduceTasks[sched]:90 if r.state == state:91 result.append(r)92 return result93

94 def hasMapTaskWithState(self, state):95 for m in self.mapTasks:96 if m.state == state:97 return True98 return False99

100 def hasReduceTaskWithState(self, scheduler, state):101 for r in self.reduceTasks[sched]:102 if r.state == state:103 return True104 return False105

106 def mapPhaseCompleted(self, currentTime):107 if self.completionTimes[’map’] == -1:108 for m in self.mapTasks:

137

109 if m.state != Task.State.Completed:110 return False111 self.completionTimes[’map’] = currentTime112 return True113

114 def reducePhaseCompleted(self, currentTime):115 finished = True116 for sched in SchedulerManager.schedulerList:117 if self.completionTimes[sched] == -1:118 thisFinished = True119 for r in self.reduceTasks[sched]:120 if r.state != Task.State.Completed:121 finished = False122 thisFinished = False123 break124 if thisFinished:125 self.completionTimes[sched] = currentTime126 return finished127

128 def isCompleted(self, currentTime):129 if not self.mapPhaseCompleted(currentTime):130 return False131 if not self.reducePhaseCompleted(currentTime):132 return False133 return True134

135 def calculateCurrentTotalCost(self, node):136 cost = 0137 for mapTask in self.getMapTasks(Task.State.Completed):138 weigth = mapTask.imputSize() * self.reduceInputRatio139 u = mapTask.node.index140 v = node.index141 hopDist = self.cluster.hopDistance(u, v)142 cost += weigth * hopDist143 return cost144

145 @staticmethod146 def stdev(x):147 mean = statistics.mean(x)148 if len(x) == 1:149 return None150 s = sum([(val - mean)**2 for val in x])151 l = (len(x) - 1)152 return math.sqrt(s/l)153

154 @staticmethod155 def randomNormal(mean, stdev):156 val = random.normalvariate(mean, stdev)157 return int(math.ceil(abs(val)))

138

158

159 @staticmethod160 def generateReduceInputRatio():161 ratio = random.normalvariate(Job.meanRatio, Job.stdevRatio)162 return abs(ratio)163

164 @staticmethod165 def generatePerSchedulerCounter():166 counter = 167 for sched in SchedulerManager.schedulerList:168 counter[sched] = 0169 return counter

Script B.10: slot.py

1 # from mrlogging import logtask2

3 class Slot:4

5 def __init__(self):6 self.available = True7 self.executingTask = None8 self.taskStartTime = None9 self.processStartTime = None

10

11 def assignTask(self, task, currentTime):12 self.available = False13 self.executingTask = task14 self.taskStartTime = currentTime15

16

17 class MapSlot(Slot):18

19 def updateStatus(self, currentTime):20 if self.available:21 return22 duration = self.executingTask.duration23 taskEndTime = self.taskStartTime + duration24 if taskEndTime < currentTime:25 self.executingTask.complete(currentTime)26 self.available = True27 self.executingTask = None28 self.taskStartTime = None29

30

31 class ReduceSlot(Slot):32

33 def assignTask(self, task, currentTime):

139

34 Slot.assignTask(self, task, currentTime)35

36 def updateStatus(self, currentTime):37 if self.available:38 return39 if not self.executingTask.shuffleComplete():40 return41 if not self.processStartTime:42 self.processStartTime = currentTime43 processDuration = self.executingTask.duration44 taskEndTime = self.processStartTime + processDuration45 if taskEndTime < currentTime:46 self.executingTask.complete(currentTime)47 totalTime = taskEndTime - self.taskStartTime48 self.available = True49 self.executingTask = None50 self.taskStartTime = None51 self.processStartTime = None

Script B.11: slotmanager.py

1 from slot import MapSlot2 from slot import ReduceSlot3

4 class SlotManager:5

6 def __init__(self, slotNumber, node):7 self.slots = self.generateSlots(slotNumber)8 self.node = node9

10 def hasAvailableSlots(self):11 for slot in self.slots:12 if slot.available:13 return True14 return False15

16 def assignTask(self, task, currentTime):17 for slot in self.slots:18 if slot.available:19 slot.assignTask(task, currentTime)20 task.assign(self.node, currentTime)21 return True22 return False23

24 def numberOfAvailableSlots(self):25 count = 026 for slot in self.slots:27 if slot.available:

140

28 count += 129 return count30

31 def updateStatus(self, currentTime):32 for slot in self.slots:33 slot.updateStatus(currentTime)34

35

36 class MapSlotManager(SlotManager):37

38 def generateSlots(self, n):39 slots = []40 for i in range(n):41 slots.append(MapSlot())42 return slots43

44 class ReduceSlotManager(SlotManager):45

46 def generateSlots(self, n):47 slots = []48 for i in range(n):49 slots.append(ReduceSlot())50 return slots

Script B.12: procedures.py

1 import math2 import random3 import statistics4 from job import Job5

6 def stdev(x):7 mean = statistics.mean(x)8 if len(x) == 1:9 return None

10 s = sum([(val - mean)**2 for val in x])11 l = (len(x) - 1)12 return math.sqrt(s/l)13

14 def generateJob(jobs, files, currentTime):15 randFile = random.choice(files)16 newJob = Job(randFile, currentTime)17 jobs.append(newJob)18 return -1

Script B.13: scheduler.py

141

1 class Scheduler:2

3 def __init__(self, cluster):4 self.cluster = cluster

Script B.14: fair.py

1 from scheduler import Scheduler2 from task import Task3

4 class Fair(Scheduler):5

6 Key = ’fair’7

8 def assignReduceTasks(self, pendingReduceQueue, currentTime):9 for node in self.cluster.shuffledNodes:

10 while (node.hasAvailableReduceSlots(self.Key) and11 len(pendingReduceQueue[self.Key])):12 reduceTask = pendingReduceQueue[self.Key].popleft()13 node.assignReduceTask(self.Key, reduceTask, currentTime)14

15 def queueReduceTasks(self, jobs, pendingReduceQueue, currentTime):16 for job in jobs:17 pending = job.getReduceTasks(self.Key, Task.State.Pending)18 if len(pending) == 0:19 continue20 completed = len(job.getMapTasks(Task.State.Completed))21 earlyShuffle = completed > (0.05 * job.numberMapTasks)22 if earlyShuffle:23 for reduceTask in pending:24 pendingReduceQueue[self.Key].append(reduceTask)25 reduceTask.queue(self.Key, currentTime)

Script B.15: coupling.py

1 import sys2 import math3 import operator4 from scheduler import Scheduler5 from task import Task6

7

8 class Coupling(Scheduler):9

10 Key = ’coupling’11 threshold = 3.012

142

13 def __init__(self, cluster, psi):14 Scheduler.__init__(self, cluster)15 self.psi = psi16 self.candidate = None17

18 # override ReduceTask assignment algorithm19 def assignReduceTasks(self, pendingReduceQueue, currTime):20 queue = pendingReduceQueue[self.Key]21 for v in self.cluster.shuffledNodes:22 if not len(queue):23 break24 if not v.hasAvailableReduceSlots(self.Key):25 continue26 self.heartbeatProcess(v, queue, currTime)27

28 def heartbeatProcess(self, v, queue, currTime):29 # wait scheduling30 if self.candidate == None:31 self.selectCandidate(queue)32 self.updateRankings(self.cluster)33 return False34 self.wait += 135 cycle = self.wait / len(self.cluster.nodes)36 if cycle < 3:37 for u in self.nodeRankings[cycle]:38 if u.index == v.index:39 availableSlots = u.hasAvailableReduceSlots(self.Key)40 while availableSlots and len(self.candidateTasks):41 r = self.candidateTasks.pop()42 queue.remove(r)43 u.assignReduceTask(self.Key, r, currTime)44 self.candidate = None45 return True46 else:47 availableSlots = v.hasAvailableReduceSlots(self.Key)48 while availableSlots and len(self.candidateTasks):49 r = self.candidateTasks.pop()50 queue.remove(r)51 v.assignReduceTask(self.Key, r, currTime)52 self.candidate = None53 return True54 return False55

56 # override ReduceTask queueing algorithm57 def queueReduceTasks(self, jobs, pendingReduceQueue, currentTime):58 for job in jobs:59 pending = job.getReduceTasks(self.Key, Task.State.Pending)60 if len(pending) == 0:61 continue

143

62 mapCompl = len(job.getMapTasks(Task.State.Completed)) * 1.063 x = (mapCompl / job.numberMapTasks)64 redLaunched = (job.numberReduceTasks - len(pending)) * 1.065 y = (redLaunched / job.numberReduceTasks)66 requestQueueing = x > y67 if requestQueueing:68 reduceTask = pending[0]69 pendingReduceQueue[self.Key].append(reduceTask)70 reduceTask.queue(self.Key, currentTime)71

72 def selectCandidate(self, queue):73 self.candidateTasks = []74 self.wait = 075 self.nodeRankings = []76 jobs = []77 for reduceTask in list(queue):78 jobs.append(reduceTask.job)79 maxMismatch = -sys.maxint80 minJob = None81 for job in list(set(jobs)):82 mismatch = self.calculateMismatch(job)83 if mismatch > maxMismatch:84 maxMismatch = mismatch85 minJob = job86 for reduceTask in list(queue):87 if reduceTask.job.index == minJob.index:88 self.candidateTasks.append(reduceTask)89 self.candidate = minJob90

91 def calculateMismatch(self, job):92 completedMaps = len(job.getMapTasks(Task.State.Completed))93 if completedMaps == job.numberMapTasks:94 rPend = len(job.getReduceTasks(self.Key, Task.State.Pending))95 rPend += len(job.getReduceTasks(self.Key, Task.State.Queued))96 mismatch = 4 + 1.0 / rPend97 else:98 thr = Coupling.threshold99 delta = 1 - math.exp(-job.numberReduceTasks / thr)

100 unit = delta * job.numberMapTasks / job.numberReduceTasks101 mapProgress = completedMaps / unit102 rRunning = len(job.getReduceTasks(self.Key, Task.State.Running))103 rCompl = len(job.getReduceTasks(self.Key, Task.State.Completed))104 redProgress = (rCompl + rRunning + 1) / unit105 mismatch = (mapProgress - redProgress) / job.numberReduceTasks106 return mismatch107

108 def updateRankings(self, cluster):109 for i in range(3):110 self.nodeRankings.append([])

144

111 nodes = 112 for node in cluster.nodes:113 if node.hasAvailableReduceSlots(self.Key):114 cost = self.candidate.calculateCurrentTotalCost(node)115 nodes[node.index] = cost116 count = 0117 for i in sorted(nodes.items(), key=operator.itemgetter(1)):118 index = i[0]119 cost = i[1]120 if count == 0:121 self.nodeRankings[0].append(cluster.nodes[index])122 elif count < 3:123 self.nodeRankings[1].append(cluster.nodes[index])124 else:125 self.nodeRankings[2].append(cluster.nodes[index])126 count += 1

Script B.16: flexible.py

1 import sys2 from coupling import Coupling3 from task import Task4

5

6 class Flexible(Coupling):7

8 Key = ’flexible’9

10 def __init__(self, cluster, psi):11 Coupling.__init__(self, cluster, psi)12

13 # override ReduceTask assignment algorithm14 def assignReduceTasks(self, pendingReduceQueue, currTime):15 queue = pendingReduceQueue[self.Key]16 maxNode = None17 maxJob = None18 minOmega = sys.maxint19 for v in self.cluster.shuffledNodes:20 if not len(queue):21 break22 if not v.hasAvailableReduceSlots(self.Key):23 continue24 assigned = self.heartbeatProcess(v, queue, currTime)25 if not assigned:26 jobs = []27 for reduceTask in list(queue):28 jobs.append(reduceTask.job)29 eligibleJobs = list(set(jobs))

145

30 if len(eligibleJobs) > 1:31 for job in list(set(jobs)):32 omega = self.omega(v, job)33 if (omega < minOmega):34 maxNode = v35 maxJob = job36 minOmega = omega37 if maxNode:38 maxJobTasks = []39 for reduceTask in list(queue):40 if reduceTask.job.index == maxJob.index:41 maxJobTasks.append(reduceTask)42 availableSlots = maxNode.hasAvailableReduceSlots(self.Key)43 if availableSlots and len(maxJobTasks):44 r = maxJobTasks.pop()45 if r in self.candidateTasks:46 self.candidateTasks.remove(r)47 queue.remove(r)48 maxNode.assignReduceTask(self.Key, r, currTime)49

50 def heartbeatProcess(self, v, queue, currTime):51 # wait scheduling52 if self.candidate == None:53 self.selectCandidate(queue)54 self.updateRankings(self.cluster)55 self.wait += 156 cycle = self.wait / len(self.cluster.nodes)57 if cycle < 3:58 for u in self.nodeRankings[cycle]:59 if u.index == v.index:60 availableSlots = u.hasAvailableReduceSlots(self.Key)61 while availableSlots and len(self.candidateTasks):62 r = self.candidateTasks.pop()63 queue.remove(r)64 u.assignReduceTask(self.Key, r, currTime)65 self.candidate = None66 return True67 else:68 availableSlots = v.hasAvailableReduceSlots(self.Key)69 while availableSlots and len(self.candidateTasks):70 r = self.candidateTasks.pop()71 queue.remove(r)72 v.assignReduceTask(self.Key, r, currTime)73 self.candidate = None74 return True75 return False76

77 def omega(self, v, job):78 mapCompl = len(job.getMapTasks(Task.State.Completed)) * 1.0

146

79 if not mapCompl > 0:80 return 081 x = (mapCompl / job.numberMapTasks)82 mapProgressPerRedNumber = x * job.numberReduceTasks83 cost = 084 gamma = 085 kappaSum = 086 for mapTask in job.getMapTasks(Task.State.Completed):87 weigth = mapTask.imputSize() * job.reduceInputRatio88 kappa = weigth / mapProgressPerRedNumber89 kappaSum += kappa90 u = mapTask.node91 hopDist = self.cluster.hopDistance(u.index, v.index)92 cost += kappa * hopDist93 gamma = kappaSum ** 294 result = cost / gamma95 return result

Script B.17: argparser.py

1 import sys2 import argparse3

4 def getArgParser():5 # parse command-line options and arguments using argparse module6 argparser = argparse.ArgumentParser(7 description="ReduceTask scheduling simulator for MapReduce",8 usage=" %(prog)s [options]"9 )

10 argparser.add_argument(11 "--jobs",12 metavar="N",13 help="number of submitted jobs",14 default=1,15 type=int16 )17 argparser.add_argument(18 "--nodes",19 metavar="N",20 help="number of nodes in the cluster",21 default=48,22 type=int23 )24 argparser.add_argument(25 "--redinput",26 metavar=("MN","STDEV"),27 help="ReduceTask input vs job input size ratio",28 default=[1.0, 0.5],

147

29 nargs=2,30 type=float31 )32 argparser.add_argument(33 "--psi",34 metavar="<f>",35 help="set <f> as Psi progress-function",36 default="linear"37 )38 argparser.add_argument(39 "--version",40 action="version",41 version=" %(prog)s 1.0",42 help="print version information and exit"43 )44 argparser.add_argument(45 "--nodesxrack",46 metavar="N",47 help="max number of nodes per rack",48 default=4,49 type=int50 )51 argparser.add_argument(52 "--switchports",53 metavar="N",54 help="max number of ports per switch",55 default=3,56 type=int57 )58 argparser.add_argument(59 "--fsizes",60 metavar="N",61 help="Possible sizes for input files",62 default=[2, 4, 8, 16, 32, 64, 128],63 nargs=’+’,64 type=int65 )66 return argparser

148

Índice de figuras

2.1. Panorama de las fases involucradas en el proceso KDD [23]. . 82.2. Visualizaciones interactivas producidas mediante D3.js [28]. . 102.3. Evolución de la performance de los procesadores [30]. . . . . . 112.4. Relación entre distintas secuencias de aminoácidos [49]. . . . . 142.5. Flujos migratorios entre los años 2005 y 2010 [50]. . . . . . . 162.6. Niveles de precipitación y temperatura en Buenos Aires. . . . 172.7. Flujo de datos para un job de MapRecue [56]. . . . . . . . . . 202.8. Esquema de ejecución de jobs en MapReduce [2]. . . . . . . . 212.9. Sistema de replicación de bloques en HDFS. . . . . . . . . . 262.10. Flujo de datos para un job en Hadoop [58, p. 30]. . . . . . . . 282.11. Detalle de las etapas Shuffle y Sort [58, p. 178]. . . . . . . . . 302.12. Reconfiguración de la estructura en Hadoop 2.0 con YARN. . 332.13. Diagrama de la arquitectura de Hadoop YARN [72]. . . . . . 352.14. Tipos de dependencias entre RDDs [82]. . . . . . . . . . . . . 382.15. DAG de ejecución de un job de Spark, divido en stages [82]. . 422.16. Performance para un job con múltiples iteraciones en Spark y

Hadoop (datos tomados de Zaharia et al. [76]). . . . . . . . . 442.17. Performance para un job con una única iteración en Spark y

Hadoop (datos tomados de Davidson et al. [86]). . . . . . . . 45

3.1. Ejemplo de alocación de recursos mediante Fair Scheduler [56]. 483.2. Relación entre componentes de COSHH [93]. . . . . . . . . . . 533.3. Diagrama esquemático del Coupling Scheduler [97]. . . . . . . 553.4. Atenuación del problema de sincronización [98]. . . . . . . . . 573.5. Comparación entre Coupling Scheduler y Fair Scheduler. . . . 593.6. Progreso de Map- y ReduceTasks en Coupling Scheduler [98]. 70

4.1. Esquema de topología en árbol [133]. . . . . . . . . . . . . . . 82

5.1. Valor promedio del tiempo de procesamiento de los jobs. . . . 103

149

5.2. Valor promedio y desviación estándar del tiempo total de eje-cución. . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 105

5.3. Nivel de tráfico en la red. . . . . . . . . . . . . . . . . . . . . 1065.4. Nivel de localidad de datos. . . . . . . . . . . . . . . . . . . . 1075.5. Tiempo de procesamiento de los jobs según su tamaño de input.1085.6. Impacto de la cantidad de nodos del cluster sobre la perfor-

mance, en términos de (a) flowtime y (b) makespan. . . . . . 1105.7. Impacto de la cantidad de nodos del cluster sobre (a) la ocu-

pación de red y (b) el nivel de localidad de datos. . . . . . . 1125.8. Impacto de la tasa de arribos de jobs sobre la performance,

en términos de (a) flowtime y (b) makespan. . . . . . . . . . 1145.9. Impacto de la tasa de arribos de jobs sobre (a) la ocupación

de red y (b) el nivel de localidad de datos. . . . . . . . . . . 116

150

Bibliografía

[1] Vernon Turner, John F. Gantz, David Reinsel, and Stephen Minton.The digital universe of opportunities: Rich data and the increasingvalue of the internet of things. IDC Research, 2014.

[2] Jeffrey Dean and Sanjay Ghemawat. MapReduce: Simplified DataProcessing on Large Clusters. OSDI ’04, 2004.

[3] Scott D. Boyd. Diagnostic Applications of High-Throughput DNA Se-quencing. Annual Review of Pathology: Mechanisms of Disease, 2013.

[4] Aaron Mckenna et al. The Genome Analysis Toolkit: a MapRedu-ce Framework for Analyzing Next-generation DNA Sequencing Data.Genome research 20, 2010.

[5] Kris Wetterstrand. DNA Sequencing Costs: Data from the NHGRI Ge-nome Sequencing Program (GSP). National Human Genome ResearchInstitute, 2014.

[6] Microsoft CityNext. Barcelona Realizes Vision of Innovative City Go-vernance with Cloud, Devices, and Apps. Microsoft Case Studies, 2013.

[7] Kim Gandhi, Larry O’Connell, Peter B. Lange, Josie Romualdi, andRamakrishnan Dhamodaran. Copenhagen Report. IBM’s SmarterCities Challenge, 2013.

[8] Jimmy Lin and Alek Kolcz. Large-Scale Machine Learning at Twitter.SIGMOD ’12, 2012.

[9] D. J. Patil. Building Data Science Teams. O’Reilly, 2011.

[10] Jeff Hammerbacher et al. Beautiful Data: The Stories Behind ElegantData Solutions. O’Reilly, 2009.

151

[11] DAMA International. The DAMA Guide to the Data ManagementBody of Knowledge. 1st Edition. Technics Publications, LLC, 2009.

[12] William H. Inmon. Building the Data Warehouse. 4th Edition. Wiley,2005.

[13] Ralph Kimball and Margy Ross. The Data Warehouse Toolkit. 2ndEdition. Wiley, 2002.

[14] Fay Chang, Jeffrey Dean, Sanjay Ghemawat, Wilson C. Hsieh, Debo-rah A. Wallach, Mike Burrows, Tushar Chandra, Andrew Fikes, andRobert E. Gruber. Bigtable: A Distributed Storage System for Struc-tured Data. OSDI, 2006.

[15] Giuseppe DeCandia, Deniz Hastorun, Madan Jampani, GunavardhanKakulapati, Avinash Lakshman, Alex Pilchin, Swaminathan Sivasu-bramanian, Peter Vosshall, and Werner Vogels. Dynamo: Amazon’sHighly Available Key-value Store. SOSP, 2007.

[16] Eben Hewitt. Cassandra, The Definitive Guide. 1st Edition. O’ReillyMedia, 2010.

[17] MongoDB Documentation, release 2.6.1 edition, 2014.

[18] Kristina Chodorow. MongoDB, The Definitive Guide. 2nd Edition.O’Reilly Media, 2013.

[19] Pramod J. Sadalage and Martin Fowler. NoSQL Distilled: A BriefGuide to the Emerging World of Polyglot Persistence. 1st Edition.Addison-Wesley Professional, 2012.

[20] Usama M. Fayyad, Gregory Piatetsky-Shapiro, and Padhraic Smyth.Advances in Knowledge Discovery and Data Mining. 1996.

[21] Gregory Piatetsky-Shapiro. Knowledge Discovery in Real Databases.1991.

[22] Oded Maimon and Lior Rokach. Data Mining and Knowledge Disco-very Handbook. 2nd Edition. Springer, 2010.

[23] Usama M. Fayyad, Gregory Piatetsky-Shapiro, and Padhraic Smyth.Knowledge Discovery and Data Mining: Towards a Unifying Frame-work. 1996.

[24] Xingquan Zhu and Ian Davidson. Knowledge Discovery and Data Mi-ning: Challenges and Realities. Information Science Reference, 2007.

152

[25] I. T. Jolliffe. Principal Component Analysis. 2nd Edition. Springer,2002.

[26] Jiawei Han and Micheline Kamber. Data Mining: Concepts and Tech-niques. 2nd Edition. 2006.

[27] Usama M. Fayyad, Gregory Piatetsky-Shapiro, and Padhraic Smyth.From Data Mining to Knowledge Discovery in Databases. 1996.

[28] Michael Bostock, Vadim Ogievetsky, and Jeffrey Heer. D3: Data-Driven Documents. IEEE Transactions on Visualization and ComputerGraphics, 2011.

[29] Doug Laney. 3d data management: Controlling data volume, velocity,and variety. META Group, 2001.

[30] John L. Hennessy and David A. Patterson. Computer Architecture: AQuantitative Approach. 5th Edition. 2011.

[31] David A. Patterson and John L. Hennessy. Computer Organizationand Design. 5th Edition. 2013.

[32] Mordechai Ben-Ari. Principles of Concurrent and Distributed Pro-gramming. Prentice Hall, 1990.

[33] George Coulouris, Jean Dollimore, and Tim Kindberg. DistributedSystems: Concepts and Design. 3rd Edition. Addison Wesley, 2001.

[34] W. Richard Stevens. TCP/IP Illustrated, Volume 1: The Protocols. 1stEdition. Addison Wesley, 1993.

[35] B. Pang and L. Lee. Opinion Mining and Sentiment Analysis. FnTIR,2(1-2), 2008.

[36] Ananth Mohan, Zheng Chen, and Kilian Weinberger. Web-SearchRanking with Initialized Gradient Boosted Regression Trees. JMLR:Workshop and Conference Proceedings 14, 2011.

[37] Jimmy Lin and Chris Dyer. Data-Intensive Text Processing with Ma-pReduce. Morgan and Claypool, 2010.

[38] Tom M. Mitchell. Machine Learning. McGraw-Hill, 1997.

[39] Christopher M. Bishop. Pattern Recognition and Machine Learning.Springer-Verlag, 2006.

153

[40] Ian H. Witten, Eibe Frank, and Mark A. Hall. Data Mining: Prac-tical Machine Learning Tools and Techniques. 3rd Edition. MorganKaufmann Publishers, 2011.

[41] Trevor Hastie, Robert Tibshirani, and Jerome Friedman. The Elementsof Statistical Learning: Data Mining, Inference, and Prediction. 2ndEdition. Springer, 2009.

[42] A. Halevy, P. Norvig, , and F. Pereira. The Unreasonable Effectivenessof Data. IEEE Intelligent Systems 24(2), 2009.

[43] Michele Banko and Eric Brill. Scaling to Very Very Large Corpora forNatural Language Disambiguation. ACL, 2001.

[44] T. Brants, A. Popat, P. Xu, F. Och, , and J. Dean. Large LanguageModels in Machine Translation. EMNLP, 2007.

[45] C. Dyer, A. Cordova, A. Mont, , and J. Lin. Fast, Easy, and Cheap:Construction of Statistical Machine Translation Models with MapRe-duce. StatMT Workshop, 2008.

[46] J. H. Friedman. Stochastic Gradient Boosting. Comput. Stat. DataAnal. 38, 4, 2002.

[47] Jerry Ye, Jyh-Herng Chow, Jiang Chen, and Zhaohui Zheng. StochasticGradient Boosted Distributed Decision Trees. CIKM’09, 2009.

[48] Martin Krzywinski, Jacqueline Schein, Inanc Birol, Joseph Connors,Randy Gascoyne, Doug Horsman, Steven J. Jones, and Marco A. Ma-rra. Circos: an Information Aesthetic for Comparative Genomics. Ge-nome Res, 2009.

[49] Jie Wang, Jiali Zhuang, and Sowmya Iyer et al. Sequence Featuresand Chromatin Structure Around the Genomic Regions Bound by 119Human Transcription Factors. Genome Res, 2012.

[50] Nikola Sander, Guy J. Abel, Ramon Bauer, and Johannes Schmidt.Visualising Migration Flow Data with Circular Plots. OEAW, 2014.

[51] J. McCarthy et al. LISP I Programmer’s Manual. Computation Centerand Research Laboratory of Electronics, MIT, 1960.

[52] Cristina L. Abad, Yi Lu, and Roy H. Campbell. DARE: Adaptive DataReplication for Efficient Cluster Scheduling. IEEE Cluster, 2011.

154

[53] Jian Tan, Xiaoqiao Meng, and Li Zhang. Improving ReduceTask DataLocality for Sequential MapReduce Jobs. INFOCOM ’13, 2013.

[54] Matei Zaharia, Andy Konwinski, Anthony D. Joseph, Randy Katz,and Ion Stoica. Improving MapReduce Performance in HeterogeneousEnvironments. OSDI ’08, 2008.

[55] Balaji Palanisamy, Aameek Singh, Ling Liu, and Bhushan Jain. Pur-lieus: Locality-Aware Resource Allocation for MapReduce in a Cloud.SC ’11, 2011.

[56] Matei Zaharia, Dhruba Borthakur, Joydeep Sen Sarma, Khaled El-meleegy, Scott Shenker, and Ion Stoica. Job Scheduling for Multi-user MapReduce Clusters. Technical Report No. UCB/EECS-2009-55,2009.

[57] Sanjay Ghemawat, Howard Gobioff, and Shun-Tak Leung. The GoogleFile System. 2013.

[58] TomWhite. Hadoop, the Definitive Guide. 3rd Edition. O’Reilly Mediaand Yahoo Press, 2012.

[59] Alberto O. Mendelzon and Juan M. Ale. Introducción a las Bases deDatos Relacionales. Prentice Hall, 2000.

[60] Grzegorz Malewicz, Matthew Austern, Aart Bik, James Dehnert, IlanHorn, Naty Leiser, and Grzegorz Czajkowski. Pregel: A System forLarge-Scale Graph Processing. SIGMOD, 2010.

[61] Mike Cafarella and Doug Cutting. Building Nutch: Open SourceSearch. ACM Queue, 2004.

[62] Christopher Olston, Benjamin Reed, Utkarsh Srivastava, Ravi Kumar,and Andrew Tomkins. Pig Latin: A Not-So-Foreign Language for DataProcessing. ACM SIGMOD Intl. Conf. on Management of Data, 2008.

[63] Ashish Thusoo, Joydeep Sen Sarma, Namit Jain, Zheng Shao, Pra-sad Chakka, Ning Zhang, Suresh Antony, Hao Liu, and RaghothamMurthy. Hive – A Petabyte Scale Data Warehouse Using Hadoop.ICDE, 2010.

[64] Dhruba Borthakur. The Hadoop Distributed File System: Architectureand Design. Apache Software Foundation, 2007.

[65] Andrew S. Tanenbaum. Modern Operating Systems. 3rd Edition. Pear-son, 2007.

155

[66] Minghong Lin, Li Zhang, AdamWierman, and Jian Tan. Joint Optimi-zation of Overlapping Phases in MapReduce. Performance Evaluation(Vol. 70), 2013.

[67] Naveen Grag, Sachin Jain, and Chaitanya Swamy. A Randomized Alo-rithm for Flow Shop Scheduling. Lecture Notes in Computer ScienceVolume 1738, 1999, pp 213-218, 1999.

[68] Abhishek Verma, Ludmila Cherkasova, and Roy H. Campbell. Two Si-des of a Coin: Optimizing the Schedule of Mapreduce Jobs to Minimi-ze Their Makespan and Improve Cluster Performance. MASCOTS’12,2012.

[69] Richard R. Weber. Scheduling Jobs With Stochastic Processing Re-quirements on Parallel Machines to Minimize Makespan or Flowtime.University of Cambridge, 1982.

[70] Yandong Wang, Kai Zhu, Lei Ying, Jian Tan, and Li Zhang. MapTask Scheduling in MapReduce with Data Locality: Throughput andHeavy-Traffic Optimality. INFOCOM ’13, 2013.

[71] Ganesh Ananthanarayanan, Ali Ghodsi, Scott Shenker, and Ion Stoica.Disk-Locality in Datacenter Computing Considered Irrelevant. Proc.USENIX Workshop on Hot Topics in Operating Syst. (HotOS), 2011.

[72] Vinod Kumar Vavilapalli, Arun C. Murthy, and Chris Douglas et al.Apache Hadoop YARN: Yet Another Resource Negotiator. SoCC’13,2013.

[73] Arun C. Murthy, Vinod Kumar Vavilapalli, and Doug Eadline et al.Apache Hadoop YARN: Moving beyond MapReduce and Batch Pro-cessing with Apache Hadoop 2. Addison Wesley Data and AnalyticsSeries, 2014.

[74] Michael Isard, M. Budiu, Yuan Yu, A. Birrell, and D. Fetterly. Dryad:Distributed Data-Parallel Programs for Sequential Building Blocks.Proc. Eurosys, 2007.

[75] Matei Zaharia, Mosharaf Chowdhury, and Tathagata Das et al. Fastand Interactive Analytics over Hadoop Data with Spark. USENIX,Vol. 37, No. 4, pp 45-51, 2012.

[76] Matei Zaharia, Mosharaf Chowdhury, Michael J. Franklin, Scott Shen-ker, and Ion Stoica. Spark: Cluster Computing with Working Sets. UCBerkeley, 2010.

156

[77] Benjamin Hindman, Andy Konwinski, Matei Zaharia, Ali Ghodsi, Ant-hony D. Joseph, Randy Katz, Scott Shenker, and Ion Stoica. Mesos:A Platform for Fine-Grained Resource Sharing in the Data Center.USENIX, 2011.

[78] Yuan Yu, Michael Isard, D. Fetterly, M. Budiu, U. Erlingsson, Pra-deep Kumar Gunda, and J. Currey. DryadLINQ: A System for General-purpose Distributed Data-parallel Computing Using a High-level Lan-guage. Proc. OSDI, 2008.

[79] Russel Power and Jinyang Li. Piccolo: Building Fast, Distributed Pro-grams with Partitioned Tables. Proc. OSDI, 2010.

[80] Yucheng Low, Danny Bickson, Joseph Gonzalez, Carlos Guestrin, AapoKyrola, and Joseph M. Hellerstein. Distributed GraphLab: A Frame-work for Machine Learning and Data Mining in the Cloud. Proceedingsof the VLDB Endowment, Vol. 5, No. 8, 2012.

[81] Matei Zaharia, Mosharaf Chowdhury, Tathagata Das, Ankur Dave,Justin Ma, Murphy McCauley, Michael J. Franklin, Scott Shenker,and Ion Stoica. Resilient Distributed Datasets: A Fault-Tolerant Abs-traction for In-Memory Cluster Computing. NSDI, 2012.

[82] Matei Zaharia. An Architecture for Fast and General Data Processingon Large Clusters. PhD thesis, UC Berkeley, 2014.

[83] Reynold Shi Xin, Joshua Rosen, Matei Zaharia, Michael Franklin, ScottShenker, and Ion Stoica. Shark: SQL and Rich Analytics at Scale. UCBerkeley - Tech. Report, 2012.

[84] Cliff Engle, Antonio Lupher, Reynold Xin, Matei Zaharia, HaoyuanLi, Scott Shenker, and Ion Stoica. Shark: Fast Data Analysis UsingCoarse-grained Distributed Memory. SIGMOD, 2012.

[85] Matei Zaharia, Tathagata Das, Haoyuan Li, Scott Shenker, and IonStoica. Discretized Streams: An Efficient and Fault-Tolerant Modelfor Stream Processing on Large Clusters. HotCloud, 2012.

[86] Aaron Davidson and Andrew Or. Optimizing Shuffle Performance inSpark. UC Berkeley - Tech. Report, 2013.

[87] Junyu Lai. Implementations of Iterative Algorithms in Hadoop andSpark. Master’s thesis, University of Waterloo, 2014.

157

[88] Matei Zaharia, Dhruba Borthakur, Joydeep Sen Sarma, Khaled Elme-leegy, Scott Shenker, and Ion Stoica. Delay Scheduling: A Simple Tech-nique for Achieving Locality and Fairness in Cluster Scheduling. Pro-ceedings of the 5th European Conference on Computer Systems, 2010.

[89] Apache Software Foundation. Capacity Scheduler Guide. Apache Ha-doop, 2008.

[90] Jagmohan Chauhan, Dwight Makaroff, and Winfried Grassmann. TheImpact of Capacity Scheduler Configuration Settings on MapReducejobs. CGC, 2012.

[91] M. Tim Jones. Scheduling in Hadoop: An Introduction to the Plugga-ble Scheduler Framework. IBM developerWorks, 2011.

[92] Jordà Polo, Claris Castillo, David Carrera, Yolanda Becerra, IanWhalley, Malgorzata Steinder, Jordi Torres, and Eduard Ayguadé.Resource-aware Adaptive Scheduling for MapReduce Clusters. AC-M/IFIP/USENIX 12th International Middleware Conference, 2011.

[93] Aysan Rasooli and Douglas G. Down. COSHH: A Classification andOptimization based Scheduler for Heterogeneous Hadoop Systems.Journal of Future Generation Computer Systems, 2012.

[94] Sameer Agarwal and Ion Stoica. Chronos: A Predictive Task Sche-duler for MapReduce. Tech. rep., EECS Department, University ofCalifornia, 2010.

[95] Thomas Sandholm and Kevin Lai. Dynamic Proportional Share Sche-duling in Hadoop. JSSPP, 2010.

[96] Thomas Sandholm and Kevin Lai. MapReduce Optimization UsingRegulated Dynamic Prioritization. SIGMETRICS ’09, 2009.

[97] Jian Tan, Xiaoqiao Meng, and Li Zhang. Coupling Scheduler for Ma-pReduce Hadoop. HPDC ’12, 2012.

[98] Jian Tan, Xiaoqiao Meng, and Li Zhang. Coupling Task Progress forMapReduce Resource-aware Scheduling. INFOCOM ’13, 2013.

[99] R. W. Wolff. Stochastic Modeling and Theory of Queues. Prentice-Hall,Englewood Cliffs, NJ, 1989.

[100] Jian Tan, Shicong Meng, Xiaoqiao Meng, and Li Zhang. Delay Asym-ptotics for Heavy-Tailed MapReduce Jobs. Allerton Conference ’12,2012.

158

[101] Joel Wolf, Deepak Rajan, Kirsten Hildrum, Rohit Khandekar, VibhoreKumar, Sujay Parekh, Kun-Lung Wu, and Andrey Balmin. FLEX:a Slot Allocation Scheduling Optimizer for MapReduce Workloads.Middleware ’10, 2010.

[102] A. V. Skorokhod. Basic Principles and Applications of ProbabilityTheory. Springer-Verlag, 2005.

[103] Predrag R. Jelenković, Xiaozhu Kang, and Jian Tan. Heavy-tailedLimits for Medium Size Jobs and Comparison Scheduling. Annals ofOperations Research, Volume 170, Issue 1, pp 133-159, 2009.

[104] Predrag R. Jelenković and Petar Momčilović. Large Deviation Analy-sis of Subexponential Waiting Times in a Processor Sharing Queue.Mathematics of Operations Research, Volume 28, Number 3, 2003.

[105] David G. Kendall. Stochastic Processes Occurring in the Theory ofQueues and their Analysis by the Method of the Imbedded MarkovChain. Annals of Mathematical Statistics, Volume 24, Number 3, pp338-354, 1953.

[106] Jian Tan, Xiaoqiao Meng, and Li Zhang. Performance Analysis ofCoupling Scheduler for MapReduce/Hadoop. INFOCOM ’12, 2012.

[107] Jian Tan, Xiaoqiao Meng, and Li Zhang. Delay Tails in MapReduceScheduling. SIGMETRICS ’12, 2012.

[108] Rudesindo Núñez Queija. Processor-Sharing Models for Integrated Ser-vices Networks. PhD thesis, Technische Universiteit Eindhoven, 2000.

[109] Soila Kavulya, Jiaqi Tan, Rajeev Gandhi, and Priya Narasimhan.An Analysis of Traces from a Production MapReduce Cluster. 10thIEEE/ACM International Conference on Cluster, Cloud and GridComputing, 2010.

[110] Xiaohong Zhang, Bin Hu, and Jiafu Jiang. An Optimized Algorithmfor ReduceTask Scheduling. Jurnal of Computers, Vol. 9, No. 4, 2014.

[111] Kurt Mehlhorn and Peter Sanders. Algorithms and Data Structures:The Basic Toolbox. Springer, 2008.

[112] Mohammad Hammoud, M. Suhail Rehman, and Majd F. Sakr. Center-of-Gravity ReduceTask Scheduling to Lower MapReduce Network Traf-fic. CLOUD ’12, 2012.

159

[113] Shadi Ibrahim, Hai Jin, Lu Lu, Song Wu, Bingsheng He, and Li Qi.Leen: Locality/fairness-aware Key Partitioning for MapReduce in theCloud. CloudCom ’10, pp 17-24, 2010.

[114] Mohammad Hammoud and Majd F. Sakr. Locality-Aware ReduceTaskScheduling for MapReduce. CLOUDCOM ’11, 2011.

[115] Kristi Morton, Abe Friesen, and Magdalena Balazinska. Estimatingthe Progress of MapReduce Pipelines. ICDE, 2010.

[116] Herodotos Herodotou, Harold Lim, Gang Luo, Nedyalko Borisov, FeiDong, Fatma Bilgen Cetin, and Shivnath Babu. Starfish: A Self-tuningSystem for Big Data Analytics. CIDR, 2011.

[117] Robert J. Mokken. Cliques, Clubs and Clans. Quality and Quantity,13, pp 161-173, 1979.

[118] Cyrus Derman, Gerald J. Lieberman, and Sheldon M. Ross. A Sequen-tial Stochastic Assignment Problem. Managment Science, Vol. 18, No7, 1972.

[119] Carlos E. García, David M. Prett, and Manfred Morari. Model Predic-tive Control: Theory and Practice. Automatica 25, pp 335-348, 1989.

[120] David Mayne, James B. Rawlings, Christopher V. Rao, and Pierre Sco-kaert. Constrained Model Predictive Control: Stability and Optimality.Automatica 36, pp 789-814, 2000.

[121] Zhuo Tang, Junqing Zhou, Kenli Li, and Ruixuan Li. MTSD: A TaskScheduling Algorithm for MapReduce Base on Deadline Constraints.IEEE 26th International IPDPSW, 2012.

[122] Shouvik Bardhan and Daniel A. Menascé. Queuing Network Models toPredict the Completion Time of the Map Phase of MapReduce Jobs.Computer Measurement Group Intl. Conf., 2012.

[123] Xiaohong Zhang, Zhiyong Zhong, Shengzhong Feng, Bibo Tu, and Jian-ping Fan. Improving Data Locality of MapReduce by Scheduling inHomogeneous Computing Environments. ISPA ’11, 2011.

[124] Michael Isard, Vijayan Prabhakaran, Jon Currey, Udi Wieder, KunalTalwar, and Andrew Goldberg. Quincy: Fair Scheduling for Distribu-ted Computing. SOSP ’09, 2009.

160

[125] Yuan Yu, Pradeep Kumar Gunda, and Michael Isard. DistributedAggregation for Data-parallel Computing: Interfaces and Implementa-tions. ACM Symposium on Operating Systems Principles, 2009.

[126] Zhenhua Guo, Geoffrey Fox, and Mo Zhou. Investigation Data Localityin MapReduce. CCGRID ’12, 2012.

[127] Ganesh Ananthanarayanan, Sameer Agarwal, Srikanth Kandula, Al-bert Greenberg, Ion Stoica, Duke Harlan, and Ed Harris. Scarlett: Co-ping With Skewed Popularity Content in MapReduce Clusters. ACMEuroSys, 2011.

[128] Yi Lu, Mei Wang, Balaji Prabhakar, and Flavio Bonomi. Elephant-Trap: A Low-cost Device for Identifying Large Flows. roc. IEEE Symp.High-Performance Interconnects, 2007.

[129] Yandong Wang, Jian Tan, Weikuan Yu, Li Zhang, and Xiaoqiao Meng.Preemptive Reducetask Scheduling for Fair and Fast Job Completion.ICAC ’13, 2013.

[130] Zhenhua Guo, M. Pierce, Geoffrey Fox, and Mo Zhou. Automatic TaskReorganization in MapReduce. CLUSTER2011, 2011.

[131] Biswapesh Chattopadhyay et al. Tenzing: A SQL Implementation onthe MapReduce Framework. Proceedings of VLDB Endowment, 2011.

[132] Yi Yao, Jianzhe Tai, Bo Sheng, and Ningfang Mi. Scheduling Hete-rogeneous MapReduce Jobs for Efficiency Improvement in EnterpriseClusters. IM ’13, 2013.

[133] Jenn-Wei Lin, Chien-Hung Chen, and J. Morris Chang. QoS-AwareData Replication for Data-Intensive Applications in Cloud ComputingSystems. IEEE Transactions on Cloud Computing, Vol. 1, No. 1, 2013.

[134] Malte Schwarzkopf, Derek G. Murray, and Steven Hand. The SevenDeadly Sins of Cloud Computing Research. HotCloud ’12, 2012.

[135] Arun C. Murthy. Mumak: Map-Reduce Simulator. Tech. Report, 2009.

[136] Abhishek Verma, Ludmila Cherkasova, and Roy H. Campbell. Play ItAgain, SimMR! IEEE International Conference on Cluster Computing.

[137] Kelvin Cardona, Jimmy Secretan, Michael Georgiopoulos, and Geor-gios Anagnostopoulos. A Grid Based System for Data Mining UsingMapReduce. Technical Report TR-2007-02, 2007.

161

[138] Guanying Wang, Aleksandr Khasymski, Krish K. R., and Ali R. Butt.Towards Improving MapReduce Task Scheduling Using Online Simu-lation Based Predictions. Modeling, Analysis and Simulation of Com-puter and Telecommunication Systems (MASCOTS), 2009.

[139] Guanying Wang. Evaluating MapReduce System Performance: A Si-mulation Approach. PhD thesis, Faculty of the Virginia PolytechnicInstitute and State University, 2012.

[140] Fei Teng, Lei Yu, and Frédéric Magoulès. SimMapReduce: A Simula-tor for Modeling MapReduce Framework. Fifth FTRA InternationalConference on Multimedia and Ubiquitous Engineering, 2012.

[141] Yang Liu, Maozhen Li, Nasullah K. Alham, and Suhel Hammoud.HSim: A MapReduce Simulator in Enabling Cloud Computing. Fu-ture Generation Computer Systems, 2011.

[142] Suhel Hammoud, Maozhen Li, Yang Liu, Nasullah K. Alham, and Ze-long Liu. MRSim: A Discrete Event Based MapReduce Simulator.Seventh International Conference on Fuzzy Systems and KnowledgeDiscovery, 2010.

[143] Abhishek Verma, Ludmila Cherkasova, and Roy H. Campbell. ARIA:Automatic Resource Inference and Allocation for MapReduce Environ-ments. Proc. of International Conference on Autonomic Computing(ICAC), 2011.

162