apache spark - uso - ¿Qué es una tarea en Spark? ¿Cómo ejecuta el trabajador de Spark el archivo jar?

uso de spark

Question

Después de leer algún documento en http://spark.apache.org/docs/0.8.0/cluster-overview.html , recibí una pregunta que quiero aclarar.

Toma este ejemplo de Spark:

JavaSparkContext spark = new JavaSparkContext(
new SparkConf().setJars("...").setSparkHome....);
JavaRDD<String> file = spark.textFile("hdfs://...");
// step1
JavaRDD<String> words =
file.flatMap(new FlatMapFunction<String, String>() {
public Iterable<String> call(String s) {
return Arrays.asList(s.split(" "));
}
});
// step2
JavaPairRDD<String, Integer> pairs =
words.map(new PairFunction<String, String, Integer>() {
public Tuple2<String, Integer> call(String s) {
return new Tuple2<String, Integer>(s, 1);
}
});
// step3
JavaPairRDD<String, Integer> counts =
pairs.reduceByKey(new Function2<Integer, Integer>() {
public Integer call(Integer a, Integer b) {
return a + b;
}
});
counts.saveAsTextFile("hdfs://...");

Entonces, digamos que tengo 3 nodos en clúster y el nodo 1 ejecutándose como maestro, y el programa de controlador anterior ha sido adecuadamente diseñado (por ejemplo, aplicación-prueba.jar). Así que ahora estoy ejecutando este código en el nodo maestro y creo que justo después de SparkContext el SparkContext , el archivo application-test.jar se copiará en los nodos de trabajadores (y cada trabajador creará un directorio para esa aplicación).

Así que ahora mi pregunta: ¿son los pasos 1, 2 y 3 en las tareas de ejemplo que se envían a los trabajadores? Si es así, ¿cómo lo ejecuta el trabajador? Al igual que java -cp "application-test.jar" step1 y así sucesivamente?

Answer 1

Cuando crea el SparkContext , cada trabajador inicia un ejecutor . Este es un proceso separado (JVM) y también carga su jar. Los ejecutores se conectan nuevamente a su programa de controlador. Ahora el controlador puede enviarles comandos, como flatMap , map y reduceByKey en su ejemplo. Cuando el conductor se cierra, los ejecutores se apagan.

Los RDD son como matrices grandes que se dividen en particiones, y cada ejecutor puede contener algunas de estas particiones.

Una tarea es un comando enviado desde el controlador a un ejecutor serializando su objeto Function . El ejecutor deserializa el comando (esto es posible porque ha cargado su jar) y lo ejecuta en una partición.

_{(Esta es una descripción conceptual. Estoy pasando por alto algunos detalles, pero espero que sea útil).}

Para responder a su pregunta específica: No, no se inicia un nuevo proceso para cada paso. Se inicia un nuevo proceso en cada trabajador cuando se construye SparkContext .

Answer 2

Para obtener una idea clara de cómo se crean y programan las tareas, debemos comprender cómo funciona el modelo de ejecución en Spark. En resumen, una aplicación en chispa se ejecuta en tres pasos:

1. Crear gráfico RDD
2. Crear plan de ejecución de acuerdo con el gráfico RDD. Las etapas se crean en este paso
3. Genere tareas basadas en el plan y haga que se programen entre los trabajadores

En su ejemplo de conteo de palabras, el gráfico RDD es bastante simple, es algo así:

archivo -> líneas -> palabras -> recuento por palabra -> conteo global de palabras -> salida

En base a este gráfico, se crean dos etapas. La regla de creación de escenario se basa en la idea de canalizar tantas transformaciones estrechas como sea posible. En su ejemplo, la transformación estrecha termina en conteo por palabra. Por lo tanto, obtienes dos etapas

1. archivo -> líneas -> palabras -> recuento por palabra
2. recuento global de palabras -> salida

Una vez que se resuelven las etapas, la chispa generará tareas desde las etapas. La primera etapa creará ShuffleMapTasks y la última etapa creará ResultTasks porque en la última etapa, se incluye una operación de acción para producir resultados.

La cantidad de tareas que se generarán depende de cómo se distribuyan sus archivos. Supongamos que tiene 3 tres archivos diferentes en tres nodos diferentes, la primera etapa generará 3 tareas: una tarea por partición.

Por lo tanto, no debe asignar sus pasos directamente a las tareas. Una tarea pertenece a una etapa y está relacionada con una partición.

Por lo general, el número de tareas ejecutadas para una etapa es exactamente el número de particiones del RDD final, pero como los RDD se pueden compartir (y por ShuffleMapStages tanto, ShuffleMapStages ) su número varía según el RDD / etapa compartida. Por favor refiérase a ¿Cómo funciona DAG bajo las cubiertas en RDD?