Lavorare con le Pair RDD in PySpark

Fondamenti di Big Data con PySpark

Upendra Devisetty

Science Analyst, CyVerse

Introduzione alle pair RDD in PySpark

I dati reali sono spesso coppie chiave/valore
Ogni riga è una chiave che mappa a uno o più valori
Le Pair RDD sono una struttura dati pensata per questi dataset
Pair RDD: la chiave è l'identificatore, il valore è il dato

Creare pair RDD

Due modi comuni per creare pair RDD
- Da una lista di tuple chiave-valore
- Da una RDD normale
Porta i dati in forma chiave/valore per la paired 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]))

Trasformazioni sulle pair RDD

Tutte le trasformazioni standard funzionano sulle pair RDD
Devi passare funzioni che operano su coppie chiave-valore, non su singoli elementi
Esempi di trasformazioni su paired RDD
- reduceByKey(func): combina i valori con la stessa chiave
- groupByKey(): raggruppa i valori con la stessa chiave
- sortByKey(): restituisce una RDD ordinata per chiave
- join(): unisce due pair RDD in base alla chiave

Trasformazione reduceByKey()

La trasformazione reduceByKey() combina i valori con la stessa chiave
Esegue operazioni in parallelo per ogni chiave nel dataset
È una trasformazione, non un'azione

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)]

Trasformazione sortByKey()

L'operazione sortByKey() ordina le pair RDD per chiave
Restituisce una RDD ordinata per chiave in ordine crescente o decrescente

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')]

Trasformazione groupByKey()

groupByKey() raggruppa tutti i valori con la stessa chiave nella 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']

Trasformazione join()

La trasformazione join() unisce due pair RDD sulla base della chiave

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))]

Esercitiamoci!

Fondamenti di Big Data con PySpark