Apache Spark: la cantidad de núcleos frente a la cantidad de ejecutores

193

Estoy tratando de entender la relación entre la cantidad de núcleos y la cantidad de ejecutores cuando se ejecuta un trabajo de Spark en YARN.

El entorno de prueba es el siguiente:

  • Número de nodos de datos: 3
  • Especificaciones de la máquina del nodo de datos:
    • CPU: Core i7-4790 (# de núcleos: 4, # de hilos: 8)
    • RAM: 32 GB (8 GB x 4)
    • HDD: 8 TB (2 TB x 4)
  • Red: 1 Gb

  • Versión Spark: 1.0.0

  • Versión de Hadoop: 2.4.0 (Hortonworks HDP 2.1)

  • Generar flujo de trabajo: sc.textFile -> filter -> map -> filter -> mapToPair -> reduceByKey -> map -> saveAsTextFile

  • Datos de entrada

    • Tipo: archivo de texto único
    • Tamaño: 165GB
    • Número de líneas: 454,568,833
  • Salida

    • Número de líneas después del segundo filtro: 310,640,717
    • Número de líneas del archivo de resultados: 99,848,268
    • Tamaño del archivo de resultados: 41 GB

El trabajo se ejecutó con las siguientes configuraciones:

  1. --master yarn-client --executor-memory 19G --executor-cores 7 --num-executors 3 (Ejecutores por nodo de datos, use tanto como núcleos)

  2. --master yarn-client --executor-memory 19G --executor-cores 4 --num-executors 3 (# de núcleos reducidos)

  3. --master yarn-client --executor-memory 4G --executor-cores 2 --num-executors 12 (menos núcleo, más ejecutor)

Tiempos transcurridos:

  1. 50 min 15 seg

  2. 55 min 48 seg

  3. 31 min 23 seg

Para mi sorpresa, (3) fue mucho más rápido.
Pensé que (1) sería más rápido, ya que habría menos comunicación entre ejecutores al barajar.
Aunque el número de núcleos de (1) es menor que (3), el número de núcleos no es el factor clave ya que 2) funcionó bien.

(Se agregaron los siguientes después de la respuesta de pwilmot).

Para la información, la captura de pantalla del monitor de rendimiento es la siguiente:

  • Resumen del nodo de datos de Ganglia para (1): el trabajo comenzó a las 04:37.

Resumen de nodo de datos de ganglios para (1)

  • Resumen de nodo de datos de Ganglia para (3): el trabajo comenzó a las 19:47. Por favor ignore el gráfico antes de ese momento.

Resumen de nodo de datos de ganglios para (3)

El gráfico se divide aproximadamente en 2 secciones:

  • Primero: desde el principio para reducir ByKey: CPU intensiva, sin actividad de red
  • Segundo: después de reduceByKey: CPU baja, se realiza la E / S de red.

Como muestra el gráfico, (1) puede usar tanta potencia de CPU como se le dio. Por lo tanto, podría no ser el problema del número de subprocesos.

¿Cómo explicar este resultado?

zeodtr
fuente
2
Ahora sospecho que GC ... De hecho, en Spark UI el tiempo total empleado para GC es más largo en 1) que en 2).
zeodtr
¿Por qué no intentaste 3) con 19G? ¿Podría ser que confinar a los trabajadores en 4G reduzca el efecto NUMA que algunas personas tienen? es decir, su 4G se encuentra en uno de los 2 núcleos asignados a su flujo de trabajo y, por lo tanto, hay una menor desaceleración de E / S, lo que conduce a un mejor rendimiento general. De lo contrario, creo que una pregunta principal es: ¿cuántos núcleos / hilo pueden usar un solo ejecutor en un trabajador? (Uno solo puede especificar el número total de núcleos para un trabajador, no a la granularidad del ejecutor)
Bacon
44
Por cierto, acabo de comprobar el código en core / src / main / scala / org / apache / spark / deploy / worker / ExecutorRunner.scala y parece que 1 ejecutor = 1 hilo del trabajador.
Tocino
un poco tarde, pero aquí hay una publicación sobre cloudera sobre este tema: blog.cloudera.com/blog/2015/03/…
Orelus
1
Por cierto, encontré esta información en una plataforma de diapositivas cloudera slideshare.net/cloudera/… , que explica un poco sobre la toma de decisiones en ejecutores, núcleos y memoria
Manish Sahni

Respuestas:

58

Con el objetivo de hacer todo esto un poco más concreto, aquí hay un ejemplo trabajado de configuración de una aplicación Spark para usar la mayor cantidad posible del clúster: imagine un clúster con seis nodos que ejecutan NodeManagers, cada uno equipado con 16 núcleos y 64 GB de memoria . Las capacidades de NodeManager, yarn.nodemanager.resource.memory-mb y yarn.nodemanager.resource.cpu-vcores, probablemente deberían establecerse en 63 * 1024 = 64512 (megabytes) y 15 respectivamente. Evitamos asignar el 100% de los recursos a contenedores YARN porque el nodo necesita algunos recursos para ejecutar el sistema operativo y los demonios Hadoop. En este caso, dejamos un gigabyte y un núcleo para estos procesos del sistema. Cloudera Manager ayuda contabilizando estos y configurando estas propiedades YARN automáticamente.

El primer impulso probable sería utilizar --num-ejecutors 6 --executor-cores 15 --executor-memory 63G . Sin embargo, este es el enfoque incorrecto porque:

63GB + la sobrecarga de memoria del ejecutor no cabe dentro de la capacidad de 63GB de los NodeManagers. El maestro de aplicaciones ocupará un núcleo en uno de los nodos, lo que significa que no habrá espacio para un ejecutor de 15 núcleos en ese nodo. 15 núcleos por ejecutor pueden conducir a un mal rendimiento de E / S HDFS.

Una mejor opción sería usar --num-ejecutors 17 --executor-cores 5 --executor-memory 19G . ¿Por qué?

Esta configuración da como resultado tres ejecutores en todos los nodos, excepto el que tiene AM, que tendrá dos ejecutores. --executor-memory se obtuvo como (63/3 ejecutores por nodo) = 21. 21 * 0.07 = 1.47. 21 - 1.47 ~ 19.

La explicación se dio en un artículo en el blog de Cloudera, How-to: Tune Your Apache Spark Jobs (Part 2) .

DzOrdre
fuente
1
"Esta configuración da como resultado tres ejecutores en todos los nodos, excepto el que tiene AM, que tendrá dos ejecutores". ¿Qué significa esto con respecto a "--executor-cores 5"?
derek
Significa que cada ejecutor usa 5 núcleos. Cada nodo tiene 3 ejecutores, por lo tanto, utiliza 15 núcleos, excepto que uno de los nodos también ejecutará el maestro de aplicaciones para el trabajo, por lo que solo puede alojar 2 ejecutores, es decir, 10 núcleos en uso como ejecutores.
Davos
Bien explicado: tenga en cuenta que esto se aplica a yarn.scheduler.capacity.resource-calculatordeshabilitado, que es el valor predeterminado. Esto se debe a que, de manera predeterminada, se programa por memoria y no por CPU.
YoYo
1
Más ejecutores pueden conducir a un mal rendimiento de E / S HDFS. Entonces, si no estoy usando HDFS, ¿en ese caso puedo usar más de 5 núcleos por ejecutor?
Darshan
Pensé que el maestro de aplicaciones se ejecuta en cada nodo. Según lo anterior, lo que significa que solo habría 1 Application Master para ejecutar el trabajo. ¿Es eso correcto?
Roshan Fernando
15

A medida que ejecuta su aplicación spark sobre HDFS, según Sandy Ryza

Me di cuenta de que el cliente HDFS tiene problemas con toneladas de hilos concurrentes. Una suposición aproximada es que, como máximo, cinco tareas por ejecutor pueden lograr un rendimiento de escritura completo, por lo que es bueno mantener el número de núcleos por ejecutor por debajo de ese número.

Por lo tanto, creo que su primera configuración es más lenta que la tercera debido al mal rendimiento de E / S de HDFS

tgbaggio
fuente
11

No he jugado con esta configuración, así que esto es solo especulación, pero si pensamos en este problema como núcleos e hilos normales en un sistema distribuido, en su clúster puede usar hasta 12 núcleos (máquinas 4 * 3) y 24 hilos (8 * 3 máquinas). En los primeros dos ejemplos, le está dando a su trabajo una cantidad justa de núcleos (espacio de cómputo potencial), pero la cantidad de subprocesos (trabajos) para ejecutar en esos núcleos es tan limitada que no puede utilizar gran parte de la potencia de procesamiento asignada y, por lo tanto, el trabajo es más lento a pesar de que hay más recursos de cálculo asignados.

Usted menciona que su preocupación estaba en el paso de barajar: aunque es bueno limitar la sobrecarga en el paso de barajar, generalmente es mucho más importante utilizar la paralelización del clúster. Piense en el caso extremo: un programa de un solo subproceso con cero aleatorio.

pwilmot
fuente
Gracias por tu respuesta Pero sospecho que el número de hilos no es el problema principal. He agregado la captura de pantalla de monitoreo. Como muestra el gráfico, 1) puede usar tanta potencia de CPU como se le dio.
zeodtr
1
@zeodtr pwilmot es correcto: necesita 2-4 tareas MÍNIMO para utilizar todo el potencial de sus núcleos. Digamos que esto era: generalmente uso al menos 1000 particiones para mi clúster de 80 núcleos.
samthebest
@samthebest Lo que quiero saber es la razón de la diferencia de rendimiento entre 1) y 3). Cuando veo la interfaz de usuario de Spark, ambas ejecutan 21 tareas en paralelo en la sección 2. (por qué 21 en lugar de 24 en el caso de 3) se desconoce por ahora) Pero, las tareas para 3) simplemente se ejecutan más rápido.
zeodtr
10

Respuesta corta : creo que tgbaggio tiene razón. Alcanza los límites de rendimiento de HDFS en sus ejecutores.

Creo que la respuesta aquí puede ser un poco más simple que algunas de las recomendaciones aquí.

La pista para mí está en el gráfico de red del clúster. Para la ejecución 1, la utilización es constante a ~ 50 M bytes / s. Para la ejecución 3, la utilización constante se duplica, alrededor de 100 M bytes / s.

Desde la publicación del blog cloudera compartida por DzOrd , puede ver esta importante cita:

Me di cuenta de que el cliente HDFS tiene problemas con toneladas de hilos concurrentes. Una suposición aproximada es que, como máximo, cinco tareas por ejecutor pueden lograr un rendimiento de escritura completo, por lo que es bueno mantener el número de núcleos por ejecutor por debajo de ese número.

Entonces, hagamos algunos cálculos para ver qué rendimiento esperamos si eso es cierto.


Ejecute 1: 19 GB, 7 núcleos, 3 ejecutores

  • 3 ejecutores x 7 hilos = 21 hilos
  • Con 7 núcleos por ejecutor, esperamos un IO limitado a HDFS (máximo a ~ 5 núcleos)
  • rendimiento efectivo ~ = 3 ejecutores x 5 hilos = 15 hilos

Ejecutar 3: 4 GB, 2 núcleos, 12 ejecutores

  • 2 ejecutores x 12 hilos = 24 hilos
  • 2 núcleos por ejecutor, por lo que el rendimiento de hdfs está bien
  • rendimiento efectivo ~ = 12 ejecutores x 2 hilos = 24 hilos

Si el trabajo está 100% limitado por concurrencia (el número de subprocesos). Esperaríamos que el tiempo de ejecución esté perfectamente inversamente correlacionado con el número de hilos.

ratio_num_threads = nthread_job1 / nthread_job3 = 15/24 = 0.625
inv_ratio_runtime = 1/(duration_job1 / duration_job3) = 1/(50/31) = 31/50 = 0.62

Entonces ratio_num_threads ~= inv_ratio_runtime, y parece que tenemos una red limitada.

Este mismo efecto explica la diferencia entre la ejecución 1 y la ejecución 2.


Run 2: 19 GB, 4 núcleos, 3 ejecutores

  • 3 ejecutores x 4 hilos = 12 hilos
  • con 4 núcleos por ejecutor, ok IO a HDFS
  • rendimiento efectivo ~ = 3 ejecutores x 4 hilos = 12 hilos

Comparando el número de hilos efectivos y el tiempo de ejecución:

ratio_num_threads = nthread_job2 / nthread_job1 = 12/15 = 0.8
inv_ratio_runtime = 1/(duration_job2 / duration_job1) = 1/(55/50) = 50/55 = 0.91

No es tan perfecto como la última comparación, pero aún vemos una caída similar en el rendimiento cuando perdemos hilos.

Ahora para el último bit: ¿por qué es el caso de que obtengamos un mejor rendimiento con más hilos, esp. más hilos que el número de CPU?

Una buena explicación de la diferencia entre paralelismo (lo que obtenemos al dividir los datos en múltiples CPU) y la concurrencia (lo que obtenemos cuando usamos múltiples hilos para trabajar en una sola CPU) se proporciona en esta gran publicación de Rob Pike: Concurrencia No es paralelismo .

La explicación breve es que si un trabajo de Spark está interactuando con un sistema de archivos o una red, la CPU pasa mucho tiempo esperando la comunicación con esas interfaces y no pasa mucho tiempo "haciendo trabajo". Al dar a esas CPU más de 1 tarea para trabajar a la vez, pasan menos tiempo esperando y más tiempo trabajando, y se ve un mejor rendimiento.

turtlemonvh
fuente
1
Explicación interesante y convincente, me pregunto si se imaginó que el ejecutor tiene un límite de 5 tareas para lograr el máximo rendimiento.
Dat Nguyen
Entonces, el número 5 no es algo que se me ocurrió: solo noté signos de cuello de botella de E / S y salí en busca de dónde pueden provenir esos cuellos de botella.
turtlemonvh
8

De los excelentes recursos disponibles en la página del paquete Sparklyr de RStudio :

DEFINICIONES DE CHISPA :

Puede ser útil proporcionar algunas definiciones simples para la nomenclatura de Spark:

Nodo : un servidor

Nodo de trabajo : un servidor que forma parte del clúster y está disponible para ejecutar trabajos de Spark

Nodo maestro : el servidor que coordina los nodos de trabajo.

Ejecutor : Una especie de máquina virtual dentro de un nodo. Un nodo puede tener múltiples ejecutores.

Driver Node : el nodo que inicia la sesión de Spark. Por lo general, este será el servidor donde se encuentra sparklyr.

Controlador (Ejecutor) : el nodo del controlador también aparecerá en la lista de Ejecutores.

d8aninja
fuente
1

Hay un pequeño problema en las dos primeras configuraciones, creo. Los conceptos de hilos y núcleos como a continuación. El concepto de subprocesos es que si los núcleos son ideales, utilice ese núcleo para procesar los datos. Por lo tanto, la memoria no se utiliza por completo en los primeros dos casos. Si desea realizar una evaluación comparativa de este ejemplo, elija las máquinas que tienen más de 10 núcleos en cada máquina. Luego haz el punto de referencia.

Pero no dé más de 5 núcleos por ejecutor, habrá un cuello de botella en el rendimiento de E / S.

Por lo tanto, las mejores máquinas para hacer este marcado de banco podrían ser nodos de datos que tienen 10 núcleos.

Especificaciones de la máquina del nodo de datos: CPU: Core i7-4790 (# de núcleos: 10, # de hilos: 20) RAM: 32GB (8GB x 4) HDD: 8TB (2TB x 4)

estrella solitaria
fuente
0

Creo que una de las principales razones es la localidad. El tamaño del archivo de entrada es 165G, los bloques relacionados del archivo ciertamente distribuidos en múltiples DataNodes, más ejecutores pueden evitar la copia de la red.

Intente establecer el número de bloques iguales del ejecutor, creo que puede ser más rápido.

zwb
fuente