Programmation avancée Spark

Spark contient deux types différents de variables partagées - l'un est broadcast variables et le second est accumulators.

• Broadcast variables - utilisé pour distribuer efficacement de grandes valeurs.

• Accumulators - utilisé pour agréger les informations d'une collection particulière.

Variables de diffusion

Les variables de diffusion permettent au programmeur de conserver une variable en lecture seule en cache sur chaque machine plutôt que d'en envoyer une copie avec les tâches. Ils peuvent être utilisés, par exemple, pour donner à chaque nœud, une copie d'un grand ensemble de données d'entrée, de manière efficace. Spark tente également de distribuer des variables de diffusion à l'aide d'algorithmes de diffusion efficaces pour réduire les coûts de communication.

Les actions Spark sont exécutées à travers un ensemble d'étapes, séparées par des opérations de «mélange» distribuées. Spark diffuse automatiquement les données communes nécessaires aux tâches à chaque étape.

Les données diffusées de cette manière sont mises en cache sous forme sérialisée et sont désérialisées avant d'exécuter chaque tâche. Cela signifie que la création explicite de variables de diffusion n'est utile que lorsque les tâches à travers plusieurs étapes nécessitent les mêmes données ou lorsque la mise en cache des données sous forme désérialisée est importante.

Les variables de diffusion sont créées à partir d'une variable v en appelant SparkContext.broadcast(v). La variable de diffusion est un wrapper autourv, et sa valeur est accessible en appelant le valueméthode. Le code donné ci-dessous montre ceci -

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))

Output -

broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

Une fois la variable de diffusion créée, elle doit être utilisée à la place de la valeur v dans toutes les fonctions exécutées sur le cluster, de sorte que vn'est pas expédié aux nœuds plus d'une fois. De plus, l'objetv ne doit pas être modifié après sa diffusion, afin de garantir que tous les nœuds obtiennent la même valeur de la variable de diffusion.

Accumulateurs

Les accumulateurs sont des variables qui ne sont «ajoutées» que par une opération associative et qui peuvent donc être efficacement prises en charge en parallèle. Ils peuvent être utilisés pour implémenter des compteurs (comme dans MapReduce) ou des sommes. Spark prend en charge nativement les accumulateurs de types numériques et les programmeurs peuvent ajouter la prise en charge de nouveaux types. Si des accumulateurs sont créés avec un nom, ils seront affichés dansSpark’s UI. Cela peut être utile pour comprendre la progression des étapes en cours d'exécution (REMARQUE - ce n'est pas encore pris en charge dans Python).

Un accumulateur est créé à partir d'une valeur initiale v en appelant SparkContext.accumulator(v). Les tâches en cours d'exécution sur le cluster peuvent ensuite y être ajoutées à l'aide duaddou l'opérateur + = (en Scala et Python). Cependant, ils ne peuvent pas lire sa valeur. Seul le programme pilote peut lire la valeur de l'accumulateur, en utilisant sonvalue méthode.

Le code donné ci-dessous montre un accumulateur utilisé pour additionner les éléments d'un tableau -

scala> val accum = sc.accumulator(0) 
 
scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x)

Si vous voulez voir la sortie du code ci-dessus, utilisez la commande suivante -

scala> accum.value

Production

res2: Int = 10

Opérations RDD numériques

Spark vous permet d'effectuer différentes opérations sur des données numériques, à l'aide de l'une des méthodes API prédéfinies. Les opérations numériques de Spark sont implémentées avec un algorithme de streaming qui permet de créer le modèle, un élément à la fois.

Ces opérations sont calculées et renvoyées sous forme de StatusCounter objet en appelant status() méthode.

Voici une liste des méthodes numériques disponibles dans StatusCounter.

S. Non	Méthodes et signification
1	count() Nombre d'éléments dans le RDD.
2	Mean() Moyenne des éléments du RDD.
3	Sum() Valeur totale des éléments dans le RDD.
4	Max() Valeur maximale parmi tous les éléments du RDD.
5	Min() Valeur minimale parmi tous les éléments du RDD.
6	Variance() Variance des éléments.
sept	Stdev() Écart-type.

Si vous souhaitez n'utiliser qu'une de ces méthodes, vous pouvez appeler la méthode correspondante directement sur RDD.