Travaux pratiques - Introduction à Spark et Scala

(la version précédente de cette séance, utilisant l’API RDD)

Références externes utiles :

Pour les séances de travaux pratiques, Spark est installé sur un système d’exploitation linux. Si vous n’êtes pas familier avec les commandes linux, il est utile de consulter au préalable ce site.

Les exemples ci-dessous reprennent, en partie, ceux des documentations en ligne de Spark.

Spark : concepts de base avec exemples

Dans les salles de travaux pratiques (TP), lors du démarrage de l’ordinateur, choisissez la configuration opensuze. Après l’ouverture de session, ouvrez une fenêtre de type terminal. Dans cette fenêtre vous pouvez entrer des commandes linux qui sont traitées par un interpréteur de commandes spécifiques (bash dans le cas particulier des salles de TP, terme générique shell).

Dans les salles, Spark est installé directement dans linux (distribution opensuze) et, dans la suite, certaines commandes système sont spécifiques à cette configuration. Si vous installez Spark d’une autre façon (avec Cloudera, par exemple), il vous faudra adapter les commandes système.

Pour réaliser le projet de cette unité d’enseignement, il vous sera nécessaire d’installer Spark sur un ordinateur auquel vous avez accès en permanence. Suivez ces instructions d’installation de Spark et signalez les éventuelles difficultés rencontrées (après avoir quand même cherché vous-même des solutions dans des forums sur le web).

Il est possible d’utiliser Spark à partir des langages Java, Scala, Python ou (dans une moindre mesure) R
  • à travers une interface en ligne de commandes (en Scala ou Python), pratique aussi bien pour des tests interactifs que pour l’étape de mise au point d’une nouvelle application,
  • en écrivant (et en exécutant ensuite) un programme (voir la séance de TP suivante).

Certaines librairies ayant été développées seulement en Scala, une application écrite en Java ou en Python doit les appeler pour employer leurs fonctionnalités. Dans ces séances de TP nous nous servirons exclusivement de Scala.

Lancement de l’interpréteur de commandes en Scala et opérations simples

Vérifiez la présence du fichier LICENSE (LICENSE avec “S” car en anglais) et copiez-le dans un répertoire tpintro à créer (attention, $ est l’invite de commandes linux, il ne faut pas l’entrer au clavier) :

$ ls /opt/spark/LICENSE
$ mkdir tpintro
$ cd tpintro
$ cp /opt/spark/LICENSE .
$ export PATH="$PATH:/opt/spark/bin"

Si vous avez installé Spark sur votre ordinateur personnel, dans les commandes ci-dessus remplacez le répertoire /opt/spark par celui dans lequel se trouve Spark sur cet ordinateur.

Lancez ensuite dans cette fenêtre l’interpréteur de commandes spark-shell en entrant

$ spark-shell

Créez un Dataset à partir du fichier texte LICENSE en entrant dans la même fenêtre (attention, ici scala> est l’invite de commandes de l’interpréteur Spark, n’entrez pas cela sur la ligne de commande de la fenêtre terminal) :

scala> val texteLicence = spark.read.textFile("LICENSE")

Si vous avez fait une faute de frappe ou de recopie et ne comprenez pas ce qui s’affiche dans la fenêtre, appelez l’enseignant ou fermez la session Spark en appuyant simultanément sur les touches Ctrl et d du clavier. Il faudra ensuite relancer spark-shell.

spark est un objet SparkSession qui comporte un certain nombre d’informations de configuration (par ex., indique à Spark comment accéder à un cluster pour exécuter les commandes). Dans ce TP les commandes seront exécutées sur un seul nœud, qui est celui sur lequel vous entrez les commandes et sur lequel tourne le programme driver de Spark.

Visualisez le fichier LICENSE. Pour cela, il faut ouvrir une seconde fenêtre terminal. Allez ensuite dans le répertoire où se trouve le fichier LICENSE et examinez-le :

$ cd tpintro
$ more LICENSE

Avec la commande linux more il faut appuyer sur la touche espace pour passer à la page suivante. Sous linux, pour voir des explications concernant une commande comme more il faut entrer man nom_commande (par exemple man more) dans une fenêtre terminal (bash).

Dataset et DataFrame

Un Dataset est une collection distribuée de données. Il peut être vu comme une évolution conceptuelle des RDD (Resilient Distributed Datasets), historiquement la première structure de données distribuée employée par Spark. Un DataFrame est un Dataset organisé en colonnes qui portent des noms, comme les tables d’une base de données. Avec l’interface de programmation en Scala, le type DataFrame est simplement l’alias du type Dataset[Row].

Il est utile de lire ces explications préparées par les créateurs de Spark pour mieux comprendre l’intérêt de chacune des interfaces de programmation (API) RDD et Dataset / Dataframe.

Il est possible d’appliquer aux Datasets des actions, qui produisent des valeurs, et des transformations, qui produisent de nouveaux Datasets, ainsi que certaines fonctions qui n’entrent dans aucune de ces deux catégories.

Quelques actions simples :

scala> texteLicence.count() // nombre de lignes dans ce Dataset

scala> texteLicence.first() // première ligne de ce Dataset

scala> texteLicence.take(5) // 5 premières lignes de ce Dataset

scala> texteLicence.collect() // retourne une partie du contenu de ce Dataset

Question : à quoi correspond une ligne de ce Dataset ?

L’exécution d’une action comme count sur un très grand Dataset se déroule de la manière suivante : le Dataset est composé de fragments, chacun stocké sur un nœud de calcul ; chaque fragment contient un (grand) nombre de lignes ; le programme driver de Spark envoie à chaque nœud le traitement à faire ; chaque nœud de calcul exécute le traitement sur le fragment local et transmet les résultats au driver. Pour d’autres actions (comme first) seul un fragment est nécessaire donc seul un nœud est sollicité.

Quelques transformations simples :

  • Construire un Dataset comprenant seulement les lignes de texteLicence qui contiennent « Copyright », retourner un tableau avec ses 2 premières lignes :
scala> val lignesAvecCopyright = texteLicence.filter(line => line.contains("Copyright"))
scala> lignesAvecCopyright.take(2) // tableau avec les 2 premières lignes de ce *Dataset*
  • Construire un Dataset composé des longueurs des lignes de texteLicence, retourner un tableau avec ses 5 premières lignes :
scala> val longueursLignes = texteLicence.map(l => l.length)
scala> longueursLignes.take(5) // tableau avec les 5 premières lignes de ce *Dataset*

Dans ces exemples, les expressions l => l.contains("Copyright") et l => l.length sont des définitions de « fonctions anonymes » (voir les éléments de Scala plus loin). Elles indiquent comment transformer chaque ligne du Dataset ; l est ainsi implicitement assimilé à une ligne générique du Dataset. Il est possible d’utiliser le nom de votre choix à la place de l, comme ligne, line, x, toto, etc.

L’exécution d’une transformation comme map sur un très grand Dataset se déroule de la manière suivante : le Dataset est composé de fragments, chacun stocké sur un nœud de calcul ; le programme driver de Spark envoie à chaque nœud le traitement à faire ; chaque nœud de calcul exécute le traitement sur le fragment local et conserve les résultats localement.

Enchaînements de transformations et d’actions :

Question :

Combien de lignes contiennent Copyright ? Écrire la commande nécessaire dans spark-shell et vérifier dans un terminal (bash) linux.

Correction

scala> texteLicence.filter(l => l.contains("Copyright")).count()

Pour vérifier, dans le second terminal pour commandes linux (et non celui avec spark-shell), entrer :

$ cd tpintro
$ cat LICENSE | grep Copyright | wc

cat LICENSE retourne le contenu du fichier LICENSE. Le caractère | indique que la sortie de la commande qui précède ne doit pas être retourné à la console mais envoyé vers la commande qui suit (c’est un enchaînement de commandes ou pipe linux). grep Copyright retourne les lignes du fichier d’entrée (ici LICENSE envoyé par la commande précédente) qui contiennent le « motif » (pattern) Copyright. Ces lignes sont envoyées (encore |) vers la commande linux wc qui retourne le nombre total de lignes, de « mots » (séquences de caractères séparées par des espaces et assimilés) et de caractères du fichier reçu en entrée (ici, les lignes de LICENSE contenant Copyright).

Question :

Combien de caractères contient le fichier ?

Correction

scala> texteLicence.map(l => l.length).reduce((a, b) => a + b)

Important : l’évaluation est « paresseuse » : les opérations sont effectuées seulement quand un résultat doit être retourné. Par exemple, dans la séquence

scala> val texteLicence = spark.read.textFile("LICENSE")
scala> val longueursLignes = texteLicence.map(l => l.length)
scala> longueursLignes.reduce((a, b) => a + b)

les données ne sont pas chargées en mémoire après la première ligne et longueursLignes n’est pas construit immédiatement après la transformation map de la seconde ligne. Ce n’est qu’au moment où l’action reduce doit être exécutée que Spark partitionne les calculs à faire en tâches pour les différentes machines (et/ou cœurs) et chaque machine (et/ou cœur) exécute sa partie de map et de reduce, avant de retourner la réponse au programme driver (qui contrôle l’exécution).

En conséquence, lorsque spark-shell vous signale une erreur, il est possible que cette erreur ne provienne pas de la dernière instruction que vous avez écrite mais d’une instruction antérieure. Pour cela, lorsque vous mettez en place une chaîne de traitement en vous servant de spark-shell, il est utile d’entrer de temps en temps des actions simples (comme longueursLignes.take(2) ici) pour identifier plus facilement les éventuelles erreurs.

Examinons plus attentivement l’action reduce((a, b) => a + b) de l’exemple précédent. L’expression (a, b) => a + b est une définition de « fonction anonyme » (voir les éléments de Scala plus loin) qui indique comment transformer une paire de lignes du Dataset ; a et b sont ainsi implicitement assimilés à deux lignes génériques du Dataset. Il est possible d’utiliser le nom de votre choix à la place de a ou b. Le déroulement de cette action est le suivant : chaque nœud, sur le fragment local du Dataset, remplace chaque paire de lignes par leur somme, ensuite applique de nouveau cette opération jusqu’à ce qu’il obtienne une seule valeur (qui sera dans ce cas la somme des valeurs de toutes les lignes) ; ces résultats sont transmis au driver qui les additionne pour obtenir la somme globale.

Si vous souhaitez supprimer les nombreuses lignes d’information qui précèdent la réponse (vous pouvez faire cela sur votre installation de Spark, pas sur celle de la salle de TP), dans le répertoire conf copiez log4j.properties.template en log4j.properties et, dans ce dernier fichier, ligne log4j.rootCategory=INFO, remplacez INFO par WARN. Sur les ordinateurs de la salle de TP, par défaut vous n’avez pas le droit d’écrire dans ce répertoire, il vous faudra précéder les commandes de copie et d’appel de l’éditeur de sudo. Dans une fenêtre terminal (bash), à ouvrir séparément de la fenêtre Spark, entrez les commandes :

$ cd /opt/spark/conf
$ sudo cp log4j.properties.template log4j.properties
$ sudo gedit log4j.properties

Dans gedit, ligne log4j.rootCategory=INFO, remplacez INFO par WARN. Pour rappel, ci-dessus $ est le prompt système dans la fenêtre bash.

Actions

Dans les exemples précédents, count, first, take, collect et reduce sont des exemples d’actions.

Voici ci-dessous quelques actions parmi les plus utilisées. Consulter la documentation (Scala) pour une liste complète et des spécifications détaillées.

Action Signification
reduce(func) Agréger les éléments du Dataset en utilisant la fonction func (qui prend 2 arguments et retourne 1 résultat). La fonction devrait être associative et commutative pour être correctement calculée en parallèle.
collect() Retourner toutes les lignes du Dataset comme un tableau au programme driver. A utiliser seulement si le Dataset a un volume faible (par ex., après des opérations de type filter très sélectives).
count() Retourner le nombre de lignes du Dataset.
take(n) Retourner un tableau avec les n premières lignes du Dataset.
head(n) Retourner un tableau avec les n premières lignes du Dataset.
first() Retourner la première ligne du Dataset (similaire à take(1)).
show(n) Afficher les n premières lignes du Dataset sous forme de tableau.
foreach(func) Appliquer la fonction func à toutes les lignes.

Transformations et persistance

Les transformations peuvent
  • produire un Dataset à partir d’un autre Dataset : map, filter, sort, etc.
  • produire un Dataset à partir de deux Dataset : union, join, crossJoin, etc.
  • produire 0 ou plusieurs Dataset à partir d’un Dataset : flatMap.

Un exemple avec flatMap : à partir de texteLicence, obtenir un Dataset motsTexteLicence ayant comme lignes les mots des différentes lignes de texteLicence et le rendre persistant pour faciliter des traitements ultérieurs.

scala> val motsTexteLicence = texteLicence.flatMap(line => line.split(" "))
scala> motsTexteLicence.persist()

Les lignes (du Dataset texteLicence) sont divisées en mots séparés par des espaces. motsTexteLicence n’est pas calculé immédiatement (évaluation paresseuse) mais le sera au moment où un résultat devra être retourné. Avec .persist() (ou .cache()), Spark cherchera à conserver ce Dataset en mémoire une fois qu’il sera calculé.

Question :

Comptez le nombre d’occurrences de chaque mot et affichez les 10 premières lignes résultantes. Indication pour la solution : .groupByKey(identity).count() permet de compter le nombre d’occurrences de chaque mot (les mots sont groupés par leur « identité », ensuite les occurrences comptées).

Correction

scala> val motsOccurrences = motsTexteLicence.groupByKey(identity).count()
scala> motsOccurrences.show(10)

Question :

Retournez la valeur maximale et la moyenne du nombre de mots des lignes du fichier.

Correction

scala> val nombresMotsParLignes = texteLicence.map(l => l.split(" ").size)
scala> nombresMotsParLignes.persist()
scala> val lMax = nombresMotsParLignes.reduce((a, b) => if (a > b) a else b)
scala> val lAvg = nombresMotsParLignes.reduce((a, b) => a + b) / texteLicence.count()

Bien entendu, il est possible d’enchaîner directement les opérations :

scala> texteLicence.flatMap(line => line.split(" ")).groupByKey(identity).count().show(10)

Noter que dans les fonctions anonymes comme line => line.split(" ") il est possible d’utiliser une autre notation pour les lignes des Datasets correspondants, par exemple a => a.split(" ") ; le mot « line » n’a pas de signification particulière pour le programme (il rappelle simplement au programmeur la signification des lignes de ces Datasets).

Il est possible de décrire le schéma d’un Dataset (DataFrame) :

scala> motsTexteLicence.printSchema()

Dans ce cas, une seule colonne est présente et se nomme « value » (nom par défaut).

Il est possible d’obtenir un nouveau DataFrame comme résultat d’une requête SQL :

scala> motsTexteLicence.createOrReplaceTempView("mots")
scala> val apresT = spark.sql("SELECT value FROM mots WHERE value > 't'")
scala> apresT.show()

Que contient apresT ?

Voici maintenant quelques transformations parmi les plus utilisées. Consulter la documentation (Scala) pour une liste complète et des spécifications détaillées.

Transformation Signification
map(func) Retourne un nouveau Dataset obtenu en appliquant la fonction func à chaque ligne du Dataset de départ.
filter(func) Retourne un nouveau Dataset obtenu en sélectionnant les lignes de la source pour lesquelles la fonction func retourne « vrai ».
flatMap(func) Similaire à map mais chaque ligne du Dataset source peut être transformée en 0 ou plusieurs lignes ; retourne une séquence (Seq) plutôt qu’une seule ligne.
sample(withReplacement,
fraction, seed)
Retourne un Dataset contenant une fraction aléatoire fraction du Dataset auquel la transformation s’applique, avec ou sans remplacement, avec une seed pour le générateur aléatoire.
union(otherDataset) Retourne un Dataset qui est l’union des lignes du Dataset source et du Dataset argument (otherDataset).
intersection(otherDataset) Retourne un Dataset qui est l’intersection des lignes du Dataset source et du Dataset argument (otherDataset).
distinct() Retourne un Dataset qui est obtenu du Dataset source en éliminant les doublons des lignes.
groupByKey(func) Pour un Dataset, calcule la clé par func et retourne un KeyValueGroupedDataset.
join(otherDataset, joinExprs,
joinType)
Pour un Dataset jointure avec otherDataset de type joinType avec condition de jointure joinExprs.
crossJoin(otherDataset) Pour Dataset de type T et otherDataset de type U, retourne un Dataset de type (T, U) (produit cartésien).

Création de Dataset, stockage de Dataset

Il y a plusieurs possibilités pour créer un Dataset :
  1. À partir d’un RDD (Resilient Distributed Dataset) existant.
  2. À partir de données externes.
  3. Transformer un (ou plusieurs) Dataset(s) existant(s).

Plusieurs solutions permettent d’obtenir un Dataset / DataFrame à partir d’un RDD. Le plus simple est de spécifier le schéma des données du DataFrame et ensuite d’appeler la méthode .createDataFrame(rowRDD, schema) de SparkSession. Afin d’obtenir un RDD à partir d’un Dataset / DataFrame il suffit d’appeler la méthode .rdd de Dataset / DataFrame.

Spark permet de créer un Dataset à partir de toute source de données acceptée par Hadoop (fichier local, HDFS, HBase, etc.). Quelques exemples :

  • SparkSession.read.textFile() : lit le fichier donné en paramètre (sous forme d’URI) et construit un Dataset dont les lignes sont les lignes du fichier texte ;
  • SparkSession.readStream... : construction de Datasets à partir d’un flux de données.

Note

Si l’URI correspond à un fichier local, il faut s’assurer que le fichier est accessible avec le même URI aux workers. Toutes les méthodes Spark de création de Dataset à partir de fichiers peuvent fonctionner sur des répertoires, des fichiers compressés et utiliser des wildcards. Par exemple : textFile("/my/directory"), textFile("/my/directory/*.txt") et textFile("/my/directory/*.gz").

Dans la configuration actuelle de Spark dans la salle de travaux pratiques, c’est le système de fichiers (filesystem) local et non HDFS (distribué) qui est utilisé par défaut. Si vous entrez simplement .textFile("LICENSE"), Spark cherchera le fichier dans le système de fichiers par défaut. Avec le préfixe file:// on indique que le fichier à lire est sur le système de fichiers local, avec le préfixe hdfs:// on indique qu’il s’agit plutôt de HDFS. L’utilisation d’un de ces préfixes exige l’emploi du chemin absolu vers le fichier correspondant.

Un Dataset peut être sauvegardé en utilisant Dataset.write..., par exemple Dataset.write.text(path) sous format texte. path indique un répertoire du système de fichiers local (préfixe file://), HDFS (préfixe hdfs://) ou autre fichier supporté par Hadoop.

Scala : concepts de base avec exemples

Scala est à la fois un langage objet pur, dans lequel chaque valeur est un objet (contrairement à Java), et un langage fonctionnel, chaque fonction étant également une valeur (donc un objet). Pour référence et un éventuel apprentissage plus approfondi du langage :

Fonctions anonymes

Les « fonctions anonymes » sont très utiles dans la définition d’opérations simples à appliquer aux données. Les expressions vues plus haut, comme (a, b) => a + b, (a, b) => if (a > b) a else b ou l => l.split(" ").size sont des définitions de fonctions anonymes. Les deux premières fonctions ont deux arguments chacune, la troisième fonction a un seul argument. La définition (a, b) => a + b est, pour des arguments qui sont des Int, une version courte de la définition suivante :

new Function1[Int, Int, Int] {
    def apply(x: Int, y: Int): Int = x + y
}

(c’est une définition formelle, non directement interprétable, ne l’entrez pas au clavier)

Il est possible de définir des fonctions anonymes avec plus de deux paramètres, par exemple (x: Int, y: Int, z: Int) => x + y - z. Une telle fonction peut être appelée directement :

scala> ((x: Int, y: Int, z: Int) => x + y - z)(1,2,3) // exemple appel fonction anonyme

Enfin, une fonction anonyme peut n’avoir aucun paramètre, par exemple () => { System.getProperty("user.dir") }, qui peut être appelée directement :

scala> (() => { System.getProperty("user.dir") })()  // exemple appel fonction anonyme

Note

Dans les expressions vues plus tôt, comme (a, b) => a + b employée par exemple dans longueursLignes.reduce((a, b) => a + b), le type des arguments n’était pas explicitement indiqué car il était donné par le type des lignes qui composent le Dataset longueursLignes.

Question :

Séparez en mots la phrase « Stat Roma pristina nomine, nomina nuda tenemus » à l’aide d’une fonction anonyme.

Correction

scala> ((l:String) => l.split(" "))("Stat Roma pristina nomine, nomina nuda tenemus")

Listes en compréhension (Sequence Comprehensions)

Les « listes en compréhension » sont des listes dont le contenu est défini par filtrage du contenu d’une autre liste, contrairement à l’approche plus classique de la construction de liste par énumération de ses éléments.

Scala propose une notation facile pour définir des listes par compréhension : for (enumerators) yield e, où enumerators est une liste de enumerators séparés par des ”;”. Un enumerator est soit un générateur qui introduit une ou plusieurs nouvelles variables, soit un filtre. Cette construction évalue l’expression e pour chaque association (binding) générée par les enumerators et retourne une séquence de ces valeurs.

Voici un exemple :

scala> def odd(from: Int, to: Int) = for (i <- from until to if i % 2 != 0) yield i
scala> println(odd(0, 20))

L’expression for dans la définition de la fonction odd introduit une nouvelle variable i: Int qui prend ensuite toutes les valeurs générées par l’itérateur _ until _ qui passent le test if i % 2 != 0 (ce test élimine les valeurs paires). Ici l’expression e de la syntaxe générale est réduite à la variable i. L’expression for retourne donc un tableau avec les nombres impairs entre from et to - 1.