Trabajar con Pair RDDs en PySpark

Fundamentos de big data con PySpark

Upendra Devisetty

Science Analyst, CyVerse

Introducción a los pair RDDs en PySpark

Los datos reales suelen ser pares clave/valor
Cada fila es una clave y apunta a uno o más valores
Un pair RDD es una estructura para trabajar con estos datos
Pair RDD: la clave identifica y el valor es el dato

Crear pair RDDs

Dos formas comunes de crear pair RDDs
- Desde una lista de tuplas clave-valor
- Desde un RDD normal
Lleva los datos a formato clave/valor para el 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]))

Transformaciones en pair RDDs

Todas las transformaciones normales funcionan en pair RDD
Debes pasar funciones que operen sobre pares clave-valor, no elementos sueltos
Ejemplos de transformaciones en pair RDD
- reduceByKey(func): combina valores con la misma clave
- groupByKey(): agrupa valores con la misma clave
- sortByKey(): devuelve un RDD ordenado por clave
- join(): une dos pair RDDs por su clave

Transformación reduceByKey()

reduceByKey() combina valores con la misma clave
Ejecuta operaciones en paralelo por cada clave del conjunto
Es una transformación, no una acción

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

Transformación sortByKey()

sortByKey() ordena un pair RDD por clave
Devuelve un RDD ordenado por clave ascendente o descendente

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

Transformación groupByKey()

groupByKey() agrupa todos los valores con la misma clave en el 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']

Transformación join()

La transformación join() une dos pair RDDs según su clave

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

¡Vamos a practicar!

Fundamentos de big data con PySpark