Comprendre les gestionnaires de shuffle dans Spark

Question

Laissez-moi vous aider à clarifier le shuffle en profondeur et comment Spark utilise gestionnaires du brassage . Je signale des ressources très utiles :

https://trongkhoanguyenblog.wordpress.com/

https://0x0fff.com/spark-architecture-shuffle/

https://github.com/JerryLead/SparkInternals/blob/master/markdown/english/4-shuffleDetails.md

En les lisant, j'ai compris qu'il existe différents gestionnaires de shuffle. Je veux me concentrer sur deux d'entre eux : hash manager y sort manager (qui est le gestionnaire par défaut).

Pour exposer ma question, je veux partir d'une transformation très courante :

val rdd = reduceByKey(_ + _)

Cette transformation provoque agrégation côté carte et ensuite remue-ménage pour rassembler toutes les mêmes clés dans une même partition.

Mes questions sont les suivantes :

• L'agrégation Map-Side est-elle mise en œuvre en utilisant en interne une transformation mapPartition et en agrégeant ainsi toutes les mêmes clés à l'aide de la fonction combiner ou est-elle mise en œuvre avec une transformation mapPartition ? AppendOnlyMap o ExternalAppendOnlyMap ?

• Si AppendOnlyMap o ExternalAppendOnlyMap Les cartes sont utilisées pour l'agrégation, sont-elles également utilisées pour l'agrégation de données ? réduire l'agrégation latérale qui se produit dans le ResultTask ?

• Quel est le but exact de ces deux types de cartes ( AppendOnlyMap o ExternalAppendOnlyMap )?

• Sont AppendOnlyMap o ExternalAppendOnlyMap utilisé par tous les gestionnaires de brassage ou seulement par le gestionnaire de tri ?

• J'ai lu qu'après AppendOnlyMap o ExternalAppendOnlyMap sont pleins, sont déversé dans un dossier comment cela se passe-t-il exactement ?

• En utilisant le gestionnaire de brassage Sort, nous utilisons un appendOnlyMap pour agréger et combiner les enregistrements de partition, n'est-ce pas ? Ensuite, lorsque la mémoire d'exécution est remplie, nous commençons à trier la carte, à la déverser sur le disque, puis à nettoyer la carte. Ma question est la suivante : quelle est la différence entre le déversement sur le disque et le shuffle write ? Ils consistent essentiellement à créer un fichier sur le système de fichiers local, mais ils sont traités différemment, les enregistrements de Shuffle write ne sont pas placés dans la carte appendOnlyMap.

• Pouvez-vous expliquer en profondeur Que se passe-t-il lorsque reduceByKey est exécuté ? Expliquez-moi toutes les étapes à suivre pour y parvenir. Par exemple, toutes les étapes de l'agrégation côté carte, du brassage, etc.

Answer 1

Il suit la description de reduceByKey étape par étape :

1. reduceByKey appelle combineByKeyWithTag avec identity combiner et identical merge value et create value
2. combineByKeyWithClassTag crée un Aggregator et retourne ShuffledRDD . Les agrégations du côté "map" et "reduce" utilisent un mécanisme interne et ne font pas appel à l'algorithme d'agrégation. mapPartitions .
3. Agregator utilise ExternalAppendOnlyMap pour les deux combineValuesByKey ("réduction côté carte") et combineCombinersByKey ("réduire le côté réduction")
4. Les deux méthodes utilisent ExternalAppendOnlyMap.insertAllMethod
5. ExternalAppendOnlyMap garde la trace des pièces renversées et la carte actuelle en mémoire ( SizeTrackingAppendOnlyMap )
6. insertAll méthode de mise à jour de la carte en mémoire et contrôles sur l'insertion si la taille estimée de la carte actuelle dépasse le seuil. Il utilise des Spillable.maybeSpill méthode. Si le seuil est dépassé, cette méthode appelle spill comme un effet secondaire y insertAll initialise le nettoyage SizeTrackingAppendOnlyMap
7. spill appelle spillMemoryIteratorToDisk qui obtient DiskBlockObjectWriter objet du gestionnaire de blocs.

insertAll Les étapes sont appliquées à la fois pour les agrégations du côté de la carte et de la réduction avec les agrégations correspondantes. Aggregator fonctions avec une étape de brassage entre les deux.

Depuis Spark 2.0, il n'existe qu'un gestionnaire basé sur le tri : SPARK-14667