Travailler avec les Pair RDDs en PySpark

Principes fondamentaux des mégadonnées avec PySpark

Upendra Devisetty

Science Analyst, CyVerse

Introduction aux pair RDDs en PySpark

Les jeux de données réels sont souvent des paires clé/valeur
Chaque ligne est une clé associée à une ou plusieurs valeurs
Un pair RDD est une structure dédiée à ce type de données
Pair RDD : la clé identifie, la valeur contient les données

Créer des pair RDDs

Deux façons courantes de créer des pair RDDs
- À partir d’une liste de tuples clé-valeur
- À partir d’un RDD classique
Mettre les données en forme clé/valeur pour un pair RDD

my_tuple = [('Sam', 23), ('Mary', 34), ('Peter', 25)]
pairRDD_tuple = sc.parallelize(my_tuple)

my_list = ['Sam 23', 'Mary 34', 'Peter 25']
regularRDD = sc.parallelize(my_list)
pairRDD_RDD = regularRDD.map(lambda s: (s.split(' ')[0], s.split(' ')[1]))

Transformations sur les pair RDDs

Toutes les transformations classiques fonctionnent sur les pair RDDs
Il faut passer des fonctions qui opèrent sur des paires clé/valeur et non sur des éléments isolés
Exemples de transformations sur pair RDD
- reduceByKey(func) : combine les valeurs ayant la même clé
- groupByKey() : regroupe les valeurs ayant la même clé
- sortByKey() : renvoie un RDD trié par clé
- join() : joint deux pair RDDs selon leur clé

Transformation reduceByKey()

reduceByKey() combine les valeurs ayant la même clé
Exécute des opérations en parallèle pour chaque clé
C’est une transformation, pas une action

regularRDD = sc.parallelize([("Messi", 23), ("Ronaldo", 34), 
                             ("Neymar", 22), ("Messi", 24)])
pairRDD_reducebykey = regularRDD.reduceByKey(lambda x,y : x + y)
pairRDD_reducebykey.collect()

[('Neymar', 22), ('Ronaldo', 34), ('Messi', 47)]

Transformation sortByKey()

sortByKey() trie un pair RDD par clé
Renvoie un RDD trié par clé en ordre croissant ou décroissant

pairRDD_reducebykey_rev = pairRDD_reducebykey.map(lambda x: (x[1], x[0]))
pairRDD_reducebykey_rev.sortByKey(ascending=False).collect()

[(47, 'Messi'), (34, 'Ronaldo'), (22, 'Neymar')]

Transformation groupByKey()

groupByKey() regroupe toutes les valeurs ayant la même clé dans le pair RDD

airports = [("US", "JFK"),("UK", "LHR"),("FR", "CDG"),("US", "SFO")]
regularRDD = sc.parallelize(airports)
pairRDD_group = regularRDD.groupByKey().collect()
for cont, air in pairRDD_group:
  print(cont, list(air))

FR ['CDG']
US ['JFK', 'SFO']
UK ['LHR']

Transformation join()

La transformation join() joint deux pair RDDs selon leur clé

RDD1 = sc.parallelize([("Messi", 34),("Ronaldo", 32),("Neymar", 24)])
RDD2 = sc.parallelize([("Ronaldo", 80),("Neymar", 120),("Messi", 100)])

RDD1.join(RDD2).collect()

[('Neymar', (24, 120)), ('Ronaldo', (32, 80)), ('Messi', (34, 100))]

Passons à la pratique !

Principes fondamentaux des mégadonnées avec PySpark