TP : Spark
Partie I :
Cette activité nécessite une programmation dans un Jupyter Notebook
Step 1 : Télecharger la dataset « Shakespeare »
Step 2 : Ouvrir un navigateur Web et créer un nouveau notebook Jupyter Python :
Créer un nouveau notebook Python en cliquant sur Nouveau, puis sur Python 3 :
Step 3 : Lire le fichier à partir de son emplacement et le stocker sous forme de RDD en
utilisant l’opération « textFile »
Quel est le type de l’opération « textFile » ? Que signifie la variable sc ?
Vérifier que le fichier a été chargé avec succès en appelant la méthode count(), qui affiche le
nombre d'éléments dans le RDD.
Step 4 : Diviser chaque ligne en des mots :
Ensuite, on va diviser chaque ligne en un ensemble de mots. Pour diviser chaque ligne en
mots et les stocker dans un RDD appelé « words » , on utilise la méthode : « flatMap » et la
fonction anonyme « Lambda »
Step 5 : Extraire sous forme clé, valeur :
Pour chaque mot on va attribuer la valeur 1 par défaut en utilisant la fonction prédéfinie
« map »
Step 6 : Reduce : faire la somme des valeurs pour chaque mot :
La méthode reduceByKey() appelle l'expression lambda pour tous les tuples avec le même
mot. L'expression lambda a deux arguments, a et b, qui sont les valeurs de comptage dans
deux tuples.
Tasnim Abar Page 1
Step 7 : Écrire le résultat dans un fichier texte dans votre disque avec
coalesce(1).saveAsTextFile()
La méthode coalesce() combine toutes les partitions RDD en une seule partition puisque on
veut un seul fichier de sortie, et saveAsTextFile() écrit le RDD à l'emplacement spécifié.
Partie II :
Exploration des données avec Spark
Manipulation d’un Dataframe
La base de données daily_weather.csv, sera partagée sur teams.
Step 1 : Après la création d’un nouveau notebook jupyter, On va charger les données dans un
Spark DataFrame. Mais tout d'abord, on doit importer la classe SparkSession:
From [Link] import sparkSession
Ensuite, on crée un sparkSession :
spark = [Link]().appName("SparkExample").getOrCreate()
Après, on charge les données dans un DataFrame :
df = [Link]("csv").load("path", header = True,
inferSchema=True)
inferSchema = true càd, il déduit automatiquement les types de colonnes en fonction des
données. Donc il nécessite de lire les données une fois de plus pour déduire le schéma.
Pour afficher les noms des colonnes, on utilise :
[Link]
Step 2 : Visualisation des colonnes et leurs types Avec les méthodes .columns et
printShema()
1- Interpréter le résultat
2- Afficher les 5 premiers éléments avec head(5)
Step 3 : Afficher les statistiques récapitulatives.
On peut afficher les statistiques récapitulatives pour toutes les colonnes en utilisant la
méthode describe() :
Tasnim Abar Page 2
toPandas() Renvoie le contenu de ceci en DataFrame tant que Pandas [Link]
On peut également voir les statistiques récapitulatives pour une seule colonne avec
describe(nom de la colonne).show()
Step 4 :
1- Créer un nouveau Dataframe avec une colonne qui s’appelle « ratio » qui calcule le
taux de rain_duration_9am/rain_accumulation_9am
Df1 = [Link]("ratio", df[‘rain_duration_9am’]/
df[‘rain_accumulation_9am’])
2- Afficher le contenu de nouvelle colonne
[Link](’ratio’).show()
3- Quel est le maximum de rain_duration_9am qui existe dans la base (utiliser orderBy et
head)
4- Calculer le mean de la colonne « rain_accumulation_9am »
from [Link] import mean
5- De même calculer le max et le min de rain_duration_9am
6- Combien de fois le air_temp_9h est > 70
[Link](‘condition’).count()
Step 5 : Suppression des lignes avec des missing values
La colonne air_pressure_9am contient des lignes avec des valeurs manquantes, on va essayer
à supprimer ces valeurs :
Tasnim Abar Page 3
Step 5 : Calculer la corrélation entre deux colonnes.
On peut calculer la corrélation entre deux colonnes dans un DataFrame en utilisant la méthode
corr().
Pourqoui on doit calculer la correlation entre les attributs dans un dataset ?
Calculer la corrélation entre rain_accumulation_9am et rain_duration_9am :
Que peut-on conclure du résultat ?
Step 6 : Imputer les valeurs manquantes.
Au lieu de supprimer les lignes contenant des valeurs manquantes, on peut les remplacer par
la valeur moyenne de cette colonne. Tout d'abord, on doit charger la fonction avg et faire une
copie du DataFrame d'origine.
Après, on doit parcourir chaque colonne du DataFrame : calculer la valeur moyenne de cette
colonne, puis remplacer toutes les valeurs manquantes dans cette colonne par la moyenne.
La fonction agg() effectue un calcul agrégé sur le DataFrame et avg(x) spécifie de calculer la
moyenne sur la colonne x.
La fonction agg() renvoie un DataFrame, first() renvoie la première Row et [0] obtient la
première valeur.
Tasnim Abar Page 4
La dernière ligne de code utilise [Link]() pour remplacer les valeurs manquantes par la valeur
moyenne (premier argument) dans la colonne x (deuxième argument).
Utiliser describe() pour comparer les statistiques de chaque colonne ; que remarquez-vous ?
Exercice d’application (par binome)
Le but de cet exercice est de créer un modèle de classification d’une base de données choisie
par vous :
Durant cette partie vous devez :
1- Définir la dataset et l’explorer
2- Faire le preprocessing nécessaire
3- Créer votre modèle de classification et la prédiction
Vous devez utiliser les bibliothèques : [Link] et [Link]
Exercice 2 : système de recommandation avec pyspark
Les deux types de systèmes de recommandation les plus courants sont :
Le filtrage basé sur le contenu se concentre sur les attributs des éléments et vous donner des
recommandations basées sur la similitude entre eux.
Le filtrage collaboratif (CF) : produit des recommandations basées sur la connaissance de
l'attitude des utilisateurs envers les articles
[Link] prend actuellement en charge le CF, dans lequel les utilisateurs et les produits sont
décrits par un petit ensemble de facteurs latents qui peuvent être utilisés pour prédire les
entrées manquantes.
[Link] utilise l'algorithme alternating least squares (ALS) pour apprendre ces facteurs
latents. Les données doivent être dans un format spécifique pour fonctionner avec l'algorithme
de recommandation ALS de Spark !
ALS est essentiellement une approche de factorisation matricielle pour implémenter un
algorithme de recommandation dans lequel vous décomposez votre grande matrice
utilisateur/élément en facteurs d'utilisateur et en facteurs d'élément de dimension inférieure.
Exemple :
User movie Rating
1 A 5
1 B 5
2 A 5
Tasnim Abar Page 5
2 B 5
1 C 1
2 C 1
3 A 5
En se basant sur le CF, on va recommander le film B pour le user 3 car les users 1 et 2
préfèrent le film A aussi
Dans cet exercice on va essayer de modéliser cet aspect avec pyspark, pour ce faire,
sous jupyter notebook :
1- Créer un sparkSession et le démarrer
2- Importer l’algo ALS :
from [Link] import ALS
from [Link] import RegressionEvaluator
3- Charger les données de « [Link] » dans un dataframe df
4- Afficher le contenu de df
5- Supprimer les colonnes inutilisables pour cet exercice
6- Diviser la dataset en training et testing sets (0.8, 0.2)
7- Créer le modèle ALS :
Als = ALS(maxIter=5, regParam=0.01, userCol=’userId’,
itemCol=’movieId’, ratingCol=’rating’)
Quel est le rôle de maxIter et regParam ??
8- Créer maintenant le modèle en utilisant [Link](training)
9- Créer une variable pred pour tester les performances de votre modèle
10- Analyser le résultat obtenu « pred »
11- Calculer le « rmse » entre les colonnes pred et rating et analyser le résultat obtenu
12- Afficher maintenant le user 600 et les différents ids des films votés par ce user
13- Appliquer le modèle crée sur le user 600
Tasnim Abar Page 6