J-11 Calculs sur SPARK

Dans ce deuxième billet je voudrais vous présenter les opérations basiques que nous pouvons réaliser sur des Resilient Distributed Datasets (RDDs) SPARK. Ce billet fait partie d’une série des exercices courtes (1h) destinés à l’apprentissage de SPARK nommés A la conquête de SPARK sur AZURE.

Parcours du Combattant:

  1. Sélection des attributs avec « map »
  2. Application des calculs avec « reduce »
  3. Agrégation avec « reduce by key »
  4. Filtrage des lignes avec « filter »

Boite à outil du combattant:

  1. Abonnement Azure ou la Version d’essai Gratuite (avec 170€)
  2. 1H de libre devant vous
  3. Avoir déployé un Cluster SPARK avoir téléchargé ce-fichier dans un Blob Storage Azure dans la compte de stockage du Cluster (voir ce billet)

1. Sélection des attributs avec « map »

Map est une fonction que nous pouvons utiliser pour sélectionner les éléments d’un RDD sur lesquels on va appliquer des calculs. Les RDD, à la base de toutes les opération dans SPARK, sont des ensembles de donnés partitionnés avec des transformations à appliquer pour les construire. Les transformations se calculent en parallèle sur les partitions au moment d’invoquer des actions.

Nous allons continuer l’exercice précédant pour cet exemple. Ouvrez le Bloc de notes Zeppellin et exécutez le le code suivant:

blog ai3 ImportJSON J-11 Calculs sur SPARK

« art » est désormais un RDD qui contient une ligne par objet Json dans notre fichier. Vu que « art » a été produit par sqlContext il s’agit d’un type particulier de RDD nommé DataFrame. Nous allons voir les DataFrames plus en profondeur dans un billet postérieur. Pour l’instant la seule chose qui nous intéresse de savoir es que un das un DataFrame chaque ligne est défini par un objet de type « Row », ou plus précisément  org.apache.spark.sql.Row. Row fournit des fonctions pour accéder aux éléments de la ligne comme row(n) (n-ième objet de la ligne en type any) ou row.getInt(n) (n-ième objet de la ligne en type int).

Pour exemplifier l’usage de map, nous allons transformer notre RDD pour qu’il contienne dans chaque ligne seulement la quantité des mots dans l’article. Une seule ligne de code va suffire:

val words = art.map( ligne => ligne.getString(0).split(« \\W »).length )

Voici quelques élément pour mieux comprendre cette ligne de code

  • Cette sentence n’applique aucun calcul sur le RDD, elle définit juste une transformation.
  • => est un opérateur scala pour définir une fonction anonyme. Cette fonction sera appliqué a chaque ligne du RDD au moment que la transformation sera appliqué par une action.
  • « ligne » va être mappé sur chaque élément du RDD et prend donc le type org.apache.spark.sql.Row pour cet exemple.
  • Le corps de la fonction est la sentence aprés =>, donc « return ligne.getString(0).split(« \\W »).length« 
  • .getString(0) retourne la première colonne de la ligne en String. Si on voulait le titre de l’article on aura du faire .getString(2)
  • .split(« \\W ») est une fonction du  String scala (lire String java) qui retourne un array en découpant la chaine de caractères originales à chaque fois que l’expression régulière passé comme paramétré est satisfaite. Dans ce cas « \W » représente tout caractère non partie d’un mot.
  • words est désormais un RDD qui contient un « Int » par ligne à la place d’un « Row » par ligne.

Pour appliquer la transformation nous allons utiliser l’action take(n) qui va exécuter notre transformation sur les premières n lignes du RDD et les retourner sous la forme d’un array. Le contenue de l’array est affiché dans la console après exécution.

blog ai3 TakeN J-11 Calculs sur SPARK

Et nous avons donc en rouge un Array avec le nombre de mots par article 🙂

 2. Application des calculs avec « reduce »

« Reduce » est une action qui va permettre d’appliquer un calcul sur toutes les lignes de notre RDD. Reduce attend comme paramètre une fonction qui reçoit deux paramétrés du même type des lignes du RDD (Int dans notre cas) et retourne un élément du même type en résultat. Devinez donc ce qui fait la ligne de code suivante:

words.reduce((a, b) => a + b)

blog ai3 NumWords J-11 Calculs sur SPARK

Et bah oui! ça calcule la quantité total des mots dans l’ensemble des articles!!

La fonction passé comme paramètre au reduce va s’appliquer progressivement (et en parallèle) à toutes les lignes du RDD. (a, b) seront donc des lignes (Int) prises de notre RDD que seront sommés pour produire le résultat final.

3. Agrégation avec « reduce by key »

Et évidement il n’y a pas de « sum » sans « group » mais pour y parvenir il va falloir changer encore un peu notre RDD. En règle général, pour grouper il faut une ou plusieurs colonnes avec des valeur répétés pour qu’ils puissent générer des groupes. Pour y parvenir il va falloir que notre RDD soit composé par paires « pair » (key, value) et on va grouper par « key ».

Ci dessous, on modifie notre code pour calculer la quantité des documents en deux groupes: Les Grands de plus de 3000 mots et les petits qui en ont moins. Voici le résultat:

blog ai3 CountBySize J-11 Calculs sur SPARK

On voit en vert le résultat final du calcul. Et les explications en suite

  • La génération du RDD des pairs (key value) est faite avec map. La fonction anonyme définie par => reçoit un « Row » et retourne un Pair[string, Int] pour chaque ligne du RDD.
  • Le constructeur des pairs est juste (,) marqué en bleue dans l’image ci dessous.
  • Le premier carré rouge est la clé des pairs. Un « If » qui retourne « petit » ou « grand » en fonction du nombre des mots. Notez comment le « If » peut aussi être utilisé comme expression qui retourne un valeur (la classe !)
  • Le deuxième carré rouge et le nombre d’articles. Donc 1 pour chaque ligne de notre RDD de base. C’est ce montant qui va être sommé avec reduceByKey
  • On orange on voit le résultat du map appliqué aux premières dix lignes.
  • La fonction passé comme paramètre à reduceByKey va s’appliquer seulement aux valeurs des pairs.

4. Filtrage des lignes avec « filter »

Notre dernier exercice consistera à filtrer les résultat du RDD. Cette fonction reçoit comme paramètre une fonction qui pour chaque ligne retourne vrai ou faux.

blog ai3 filter J-11 Calculs sur SPARK

  •  On voit en vert la quantité des articles contenant le texte « super« 
  • A différence de map, filter (surchargé pour les DataFrames) nous oblige à utiliser une expression avec $(« colonne ») pour faire référence aux colonnes des lignes.
  • Si à la place d’un DataFrame nous aurions eu un RDD de base, l’expression aura du être écrite art.filter(ligne => ligne.getString(0).contains(« super »))

Bon super! Jusque le prochain billet! Où je vais vous présenter quelques opérations plus complexes avec SPARK.

One thought on “J-11 Calculs sur SPARK

  1. Pingback: A la conquête de SPARK sur AZURE | Le blog Ai3Le blog Ai3

Laisser un commentaire

Votre adresse de messagerie ne sera pas publiée. Les champs obligatoires sont indiqués avec *

Ce site utilise Akismet pour réduire les indésirables. En savoir plus sur comment les données de vos commentaires sont utilisées.