Configurando Spark en YARN

Habr, hola! Ayer el reunión dedicada a Apache Spark, por parte de los chicos de Rambler&Co, hubo bastantes preguntas por parte de los participantes relacionadas con la configuración de esta herramienta. Decidimos seguir sus pasos y compartir nuestra experiencia. El tema no es fácil, por eso te invitamos a compartir tu experiencia en los comentarios, tal vez nosotros también entendamos y usemos algo mal.

Una pequeña introducción a cómo usamos Spark. Tenemos un programa de tres meses. “Especialista en Big Data”, y a lo largo del segundo módulo nuestros participantes trabajan este instrumento. En consecuencia, nuestra tarea como organizadores es preparar el grupo para su uso en tal caso.

La peculiaridad de nuestro uso es que la cantidad de personas que trabajan simultáneamente en Spark puede ser igual a la de todo el grupo. Por ejemplo, en un seminario, cuando todos intentan algo al mismo tiempo y lo repiten después de nuestro profesor. Y esto no es mucho, a veces hasta 40 personas. Probablemente no haya muchas empresas en el mundo que se enfrenten a un caso de uso de este tipo.

A continuación, les diré cómo y por qué seleccionamos ciertos parámetros de configuración.

Empecemos desde el principio. Spark tiene 3 opciones para ejecutarse en un clúster: independiente, usando Mesos y usando YARN. Decidimos elegir la tercera opción porque para nosotros tenía sentido. Ya tenemos un clúster de hadoop. Nuestros participantes ya conocen bien su arquitectura. Usemos HILO.

spark.master=yarn

Más interesante. Cada una de estas 3 opciones de implementación tiene 2 opciones de implementación: cliente y clúster. Basado documentación Y varios enlaces en Internet, podemos concluir que el cliente es adecuado para el trabajo interactivo, por ejemplo, a través de Jupyter Notebook, y el clúster es más adecuado para soluciones de producción. En nuestro caso, nos interesaba el trabajo interactivo, por tanto:

spark.deploy-mode=client

En general, a partir de ahora Spark funcionará de alguna manera en YARN, pero esto no fue suficiente para nosotros. Como tenemos un programa sobre big data, a veces los participantes no tenían suficiente de lo obtenido en el marco de una distribución equitativa de los recursos. Y luego encontramos algo interesante: la asignación dinámica de recursos. En resumen, el punto es el siguiente: si tiene una tarea difícil y el clúster está libre (por ejemplo, por la mañana), usar esta opción Spark puede brindarle recursos adicionales. Allí la necesidad se calcula según una astuta fórmula. No entraremos en detalles: funciona bien.

spark.dynamicAllocation.enabled=true

Configuramos este parámetro y, al iniciar Spark, falló y no se inició. Así es, porque tuve que leerlo. documentación mas cuidadosamente. Dice que para que todo esté bien, también es necesario habilitar un parámetro adicional.

spark.shuffle.service.enabled=true

¿Por qué es necesario? Cuando nuestro trabajo ya no requiera tantos recursos, Spark debería devolverlos al fondo común. La etapa que consume más tiempo en casi cualquier tarea de MapReduce es la etapa Shuffle. Este parámetro le permite guardar los datos que se generan en esta etapa y liberar a los ejecutores en consecuencia. Y el ejecutor es el proceso que calcula todo sobre el trabajador. Tiene una determinada cantidad de núcleos de procesador y una determinada cantidad de memoria.

Este parámetro ha sido agregado. Todo parecía funcionar. Se hizo evidente que los participantes recibieron más recursos cuando los necesitaron. Pero surgió otro problema: en algún momento otros participantes se despertaron y también quisieron usar Spark, pero allí todo estaba ocupado y no estaban contentos. Se pueden entender. Empezamos a mirar la documentación. Resultó que existen otros parámetros que pueden utilizarse para influir en el proceso. Por ejemplo, si el ejecutor está en modo de espera, ¿después de qué tiempo se le pueden quitar recursos?

spark.dynamicAllocation.executorIdleTimeout=120s

En nuestro caso, si sus ejecutores no hacen nada durante dos minutos, devuélvalos al grupo común. Pero este parámetro no siempre fue suficiente. Estaba claro que la persona llevaba mucho tiempo sin hacer nada y no se liberaban recursos. Resultó que también hay un parámetro especial: después de qué tiempo seleccionar los ejecutores que contienen datos almacenados en caché. ¡Por defecto, este parámetro era infinito! Lo corregimos.

spark.dynamicAllocation.cachedExecutorIdleTimeout=600s

Es decir, si tus ejecutores no hacen nada durante 5 minutos, entrégalos al fondo común. En este modo, la velocidad de liberación y emisión de recursos para una gran cantidad de usuarios se ha vuelto decente. La cantidad de descontento ha disminuido. Pero decidimos ir más allá y limitar el número máximo de ejecutores por solicitud, esencialmente por participante del programa.

spark.dynamicAllocation.maxExecutors=19

Ahora, por supuesto, hay gente insatisfecha del otro lado: "el clúster está inactivo y solo tengo 19 ejecutores", pero ¿qué se puede hacer? Necesitamos algún tipo de equilibrio correcto. No puedes hacer felices a todos.

Y una pequeña historia más relacionada con las particularidades de nuestro caso. De alguna manera, varias personas llegaron tarde a una lección práctica y, por alguna razón, Spark no comenzó por ellos. Observamos la cantidad de recursos gratuitos; parece estar ahí. La chispa debería comenzar. Afortunadamente, en ese momento la documentación ya se había agregado a la subcorteza en algún lugar y recordamos que cuando se inicia, Spark busca un puerto para comenzar. Si el primer puerto del rango está ocupado, pasa al siguiente en orden. Si es gratis, captura. Y hay un parámetro que indica el número máximo de intentos para ello. El valor predeterminado es 16. El número es menor que el número de personas de nuestro grupo en clase. En consecuencia, después de 16 intentos, Spark se rindió y dijo que no podía empezar. Hemos corregido esta configuración.

spark.port.maxRetries=50

A continuación os hablaré de algunas configuraciones que no están muy relacionadas con las particularidades de nuestro caso.

Para iniciar Spark más rápido, se recomienda archivar la carpeta jars ubicada en el directorio de inicio SPARK_HOME y colocarla en HDFS. Entonces no perderá el tiempo cargando estos jarniks por parte de los trabajadores.

spark.yarn.archive=hdfs:///tmp/spark-archive.zip

También se recomienda utilizar kryo como serializador para una operación más rápida. Está más optimizado que el predeterminado.

spark.serializer=org.apache.spark.serializer.KryoSerializer

Y también hay un problema de larga data con Spark que a menudo falla desde la memoria. A menudo esto sucede en el momento en que los trabajadores han calculado todo y envían el resultado al conductor. Hicimos este parámetro más grande para nosotros mismos. Por defecto, es 1 GB, nosotros lo hicimos 3.

spark.driver.maxResultSize=3072

Y por último, como postre. Cómo actualizar Spark a la versión 2.1 en la distribución HortonWorks - HDP 2.5.3.0. Esta versión de HDP contiene una versión 2.0 preinstalada, pero una vez decidimos por nosotros mismos que Spark se está desarrollando de manera bastante activa y que cada nueva versión corrige algunos errores y proporciona funciones adicionales, incluida la API de Python, por lo que decidimos qué es necesario. lo que hay que hacer es una actualización.

Descargué la versión del sitio web oficial para Hadoop 2.7. Lo descomprimimos y lo guardamos en la carpeta HDP. Instalamos los enlaces simbólicos según sea necesario. Lo lanzamos, no arranca. Escribe un error muy poco claro.

java.lang.NoClassDefFoundError: com/sun/jersey/api/client/config/ClientConfig

Después de buscar en Google, descubrimos que Spark decidió no esperar hasta que naciera Hadoop y decidió usar la nueva versión de Jersey. Ellos mismos discuten entre ellos sobre este tema en JIRA. La solución fue descargar versión de camiseta 1.17.1. Coloque esto en la carpeta jars en SPARK_HOME, comprímalo nuevamente y cárguelo en HDFS.

Solucionamos este error, pero surgió uno nuevo y bastante simplificado.

org.apache.spark.SparkException: Yarn application has already ended! It might have been killed or unable to launch application master

Al mismo tiempo, intentamos ejecutar la versión 2.0; todo está bien. Intenta adivinar qué está pasando. Examinamos los registros de esta aplicación y vimos algo como esto:

/usr/hdp/${hdp.version}/hadoop/lib/hadoop-lzo-0.6.0.${hdp.version}.jar

En general, por alguna razón hdp.version no se resolvió. Después de buscar en Google, encontramos una solución. Debes ir a la configuración de YARN en Ambari y agregar un parámetro allí al sitio de hilo personalizado:

hdp.version=2.5.3.0-37

Esta magia ayudó y Spark despegó. Probamos varias de nuestras computadoras portátiles jupyter. Todo está funcionando. ¡Estamos listos para la primera lección de Spark el sábado (mañana)!

UPD. Durante la lección, salió a la luz otro problema. En algún momento, YARN dejó de proporcionar contenedores para Spark. En YARN fue necesario corregir el parámetro, que por defecto era 0.2:

yarn.scheduler.capacity.maximum-am-resource-percent=0.8

Es decir, sólo el 20% de los recursos participó en la distribución de recursos. Después de cambiar los parámetros, recargamos YARN. El problema se resolvió y el resto de los participantes también pudieron ejecutar Spark Context.

Fuente: habr.com