Pourquoi ces deux méthodes de génération de RDDs Spark ont-elles des localisations de données différentes ?

Question

J'exécute deux méthodes différentes de génération de RDDs dans une machine locale, la première étant la suivante :

rdd = sc.range(0, 100).sortBy(lambda x: x, numPartitions=10)
rdd.collect()

La deuxième façon est :

rdd = sc.parallelize(xrange(100), 10)
rdd.collect()

Mais dans mon interface utilisateur Spark, la localité des données est différente, et je ne sais pas pourquoi. Voici le résultat de la première façon, il montre Locality Level (la 5ème colonne) est ANY.

Et le résultat de la deuxième façon montre que le niveau de localité est Process_Local :

Et je lis de https://spark.apache.org/docs/latest/tuning.html Le niveau Process_Local est généralement plus rapide que le niveau Any pour le traitement.

Est-ce à cause de l'opération sortBy qui va donner lieu à un shuffle puis influencer la localité des données ? Quelqu'un peut-il me donner une explication plus claire ?

Answer 1

Vous avez raison.

Dans le premier extrait, vous créez d'abord une collection parallélisée, ce qui signifie que votre pilote demande à chaque travailleur de créer une partie de la collection. Ensuite, comme pour le tri, chaque nœud de travail doit accéder aux données des autres nœuds, les données doivent être mélangées et la localité des données est perdue.

Le deuxième extrait de code n'est en fait même pas un travail distribué.

Comme Spark utilise l'évaluation paresseuse, rien n'est fait jusqu'à ce que vous appeliez pour matérialiser les résultats, dans ce cas en utilisant la méthode collect. Les étapes de votre deuxième calcul sont effectivement

1. Distribuer l'objet de type liste du pilote vers les nœuds de travailleurs
2. Ne rien faire sur chaque nœud de travail
3. Collecte des objets distribués par les travailleurs pour créer un objet de type liste sur le conducteur.

Spark est assez intelligent pour se rendre compte qu'il n'y a aucune raison de distribuer la liste même si paralléliser est appelé. Puisque les données résident et que le calcul est effectué sur le même nœud unique, la localité des données est évidemment préservée.

EDIT : Quelques informations supplémentaires sur la façon dont Spark fait le tri.

Spark fonctionne sur le modèle MapReduce sous-jacent (le modèle de programmation, pas l'implémentation Hadoop) et le tri est implémenté comme un seul map et un reduce. Conceptuellement, sur chaque nœud de la phase map, la partie de la collection sur laquelle un nœud particulier opère est triée et écrite en mémoire. Les réducteurs extraient ensuite les données pertinentes des mappeurs, fusionnent les résultats et créent des itérateurs.

Ainsi, pour votre exemple, disons que vous avez un mappeur qui a écrit les numéros 21-34 en mémoire dans un ordre trié. Disons que le même nœud a un reducer qui est responsable des numéros 31-40. Le reducer reçoit des informations du pilote où se trouvent les données pertinentes. Les numéros 31-34 sont tirés du même nœud et les données n'ont qu'à voyager entre les threads. En revanche, les autres numéros peuvent se trouver sur n'importe quel nœud du cluster et doivent être transférés sur le réseau. Une fois que le réducteur a extrait toutes les données pertinentes des nœuds, la phase de brassage est terminée. Le réducteur fusionne maintenant les résultats (comme dans mergesort) et crée un itérateur sur la partie triée de la collection.

http://blog.cloudera.com/blog/2015/01/improving-sort-performance-in-apache-spark-its-a-double/