Operaciones avanzadas con RDDs en Spark: Uniones y Joins

Clase 10 de 25 • Curso de Fundamentos de Spark para Big Data

Resumen

¿Cómo operar con múltiples RDDs para obtener información significativa?

Operar con RDDs (Resilient Distributed Datasets) es una habilidad esencial para quienes trabajan con grandes volúmenes de datos. En esta lección, te mostramos cómo integrar y manipular varios RDDs para obtener datos significativos. Al finalizar este proceso, tendrás una comprensión más profunda de cómo gestionar estos datos de manera efectiva en un entorno distribuido.

¿Cómo importar y visualizar RDDs en Jupyter?

Comenzamos importando RDDs desde archivos CSV usando Jupyter. Para esto, asegúrate de tener los archivos CSV en tu carpeta de trabajo. Recuerda que los archivos pueden tener o no encabezados. Por ello, importamos dos archivos: uno con encabezado ("deportista.csv") y otro sin ("deportista2.csv").

deportista_olimpico_RDD = sc.textFile('path/to/deportista.csv').map(lambda l: l.split(','))
deportista_olimpico_RDD2 = sc.textFile('path/to/deportista2.csv').map(lambda l: l.split(','))

¿Cómo realizar una operación Union en RDDs?

La operación union es crucial cuando necesitas fusionar datos de dos RDDs. Spark maneja automáticamente las duplicidades al ejecutar una union. Aquí te mostramos cómo hacerlo:

deportista_olimpico_unido = deportista_olimpico_RDD.union(deportista_olimpico_RDD2)

Para validar la operación, puedes utilizar el conteo:

conteo = deportista_olimpico_unido.count()

Un conteo exitoso indica que los archivos tienen la calidad mínima necesaria.

¿Cómo realizar joins con RDDs?

Hacer un join entre RDDs te permite combinar información de diferentes fuentes. Por ejemplo, al unir deportistas con equipos olímpicos, primero debes identificar las claves comunes:

# Seleccionar la clave y el resto de las columnas necesarias
deportista_clave_valor = deportista_olimpico_RDD.map(lambda l: (l[-1], l))
equipos_olimpicos_clave_valor = equipos_olimpicos_RDD.map(lambda m: (m[0], m[-1]))

# Realizar el join
join_resultado = deportista_clave_valor.join(equipos_olimpicos_clave_valor)

¿Cómo utilizar muestras para verificar un join?

La función takeSample te permite tomar una muestra aleatoria de tus datos para revisar la calidad del join:

muestra = join_resultado.takeSample(False, 6, 42)

¿Cómo cargar y filtrar un RDD de resultados?

Al trabajar con datos como resultados de juegos olímpicos, puedes estar interesado solo en aquellos jugadores que ganaron medallas. Para esto, carga y filtra los resultados:

resultados_RDD = sc.textFile('path/to/Resultados.csv').map(lambda l: l.split(','))
resultados_ganadores = resultados_RDD.filter(lambda l: 'NA' not in l[1])

Recomendaciones finales

Al operar con grandes volúmenes de datos, es clave utilizar técnicas eficientes. A medida que desarrolles tus habilidades, explora el uso de más funciones de Spark para optimizar tus análisis. ¡No olvides seguir practicando y buscando nuevos retos para consolidar tus conocimientos!

Víctor Macedo Becerril

student•

El truco de este reto es:

Dado el join que se acaba de realizar entre deportistaOlimpicoRDD y equiposOlimpicosRDD, hay que cambiar el valor llave (el cual es equipo_id) por deportista_id ya que éste valor es quien relaciona la tabla Deportista con la tabla Resultados.

Solo después de haber realizado el cambio de valor, ya se puede hacer el join. El cambio de llaves se hace simplemente ordenando la información con un map().

(deportistaOlimpicoRDD.map(lambda l : [l[-1],l[:-1]]) # Se selecciona la última columna del RDD (equipo_id) que es valor eje y el resto de contenidos
                        .join(equiposOlimpicosRDD.map(lambda l : [l[0],l[2]])) # Se selecciona solo el id que es el valor eje y la sigla del país
                        .map(lambda l : (l[1][0][0], (l[0],l[1][0][1:],l[1][1]))) # Pone deportista_id como valor eje
                        .join(resultadoGanador.map(lambda l : [l[2],l[1]])) # Toma deportista_id y medalla
                        .takeSample(False,6,25))

mi output es:

[('54832',
  (('982', ['Carlos Jimnez Snchez', '1', '24', '201', '100'], 'ESP'),
   'Silver')),
 ('135219',
  (('944', ['Urka olnir', '2', '22', '173', '64'], 'SLO'), 'Bronze')),
 ('106724',
  (('705', ['Adrie Ard Schenk', '1', '19', '190', '90'], 'NED'), 'Gold')),
 ('112317',
  (('1096', ['Justus Ketchum Smith', '1', '26', '0', '0'], 'USA'), 'Gold')),
 ('124518',
  (('705', ['Daniel Daan van Dijk', '1', '21', '0', '0'], 'NED'), 'Gold')),
 ('42668',
  (('1096', ['Joseph Tilford Lee Joe Greene', '1', '25', '183', '70'], 'USA'),
   'Bronze'))]

Juan Pablo Reina Gutierrez

student•

Así yo le dí respuesta al reto para traernos los valores de todos los campos:

Johan Steven Arias Pérez

student•

Tener en cuenta que la función top() NO muestra los primeros valores, en realidad muestra los valores más altos. Es decir si tengo un arreglo con los valores [1,2,3,4,5] y uso top(2) me mostrará el valor 5 y 4. Si mi arreglo es de strings me mostrará primero las letras en orden alfabético seguido de los números. Ejemplo: si mi RDD es [‘a’,‘b’,‘4’,‘5’] y uso top(3) me mostrará [‘b’,‘a’,‘5’]..

Jose Daniel Gallegos Padilla

student•

hola

Hubert Ronald Mendoza Canales

student•

No sé que país sea este PER: Perú ;)

Fabio Gomez

student•

Reto con equipo id, nombre, siglas y medalla

deportistaOlimpicoRDD.map(lambda l: [l[-1], l[1]]) \
    .join(equiposOlimpicosRDD.map(lambda x: [x[0], x[2]])) \
    .join(resultadoGanador.map(lambda i: [i[2], i[1]])) \
    .takeSample(False, 5, 11)

[('507', (('Primo Baran', 'ITA'), 'Gold')), ('15', (('Ahmed Sad', 'ALG'), 'Bronze')), ('514', (('Takaji Mori', 'JPN'), 'Silver')), ('399', (('Andreas Wellinger', 'GER'), 'Silver')), ('507', (('Gianfranco Grasselli', 'ITA'), 'Bronze'))]

Juan Jose Tovar

student•

¿Cómo hago para que me queden los headers después del Join?

Joan Sebastián Muñoz Ramírez

student•

Sé que es un curso muy viejo y tal vez nadie vea mi respuesta. Pero creo que es muy diferente a la que los demás han hecho y creo que es mucho más fácil de entender.

ganadoresOlimpicos = deportistaOlimpicoRDD.map(lambda l: (l[-1], l[:-1])) \
    .join(equiposOlimpicosRDD.map(lambda l: (l[0], l[2]))) \
    .map(lambda l: (l[1][0][0], tuple([l[0]] + l[1][0][1:] + [l[1][1]]))) \
    .join(resultadoGanador.map(lambda l: (l[2], l[1])))\
    .take(6)

Iván Mauricio Jaimes Niño

student•

Modelo relacional

JAVIER SANTIAGO SALGADO

student•

Lo duro es entender que es una lista dentro de una tupla. De nuevo gracias a la comunidad Platzi, sin ellos sería imposible.

deportistaOlimpicosRDD.map(lambda l: [l[-1], l[:-1]]) \
    .join(equiposOlimpicosRDD.map(lambda x: [x[0], x[2]]))\
    .map(lambda r: (r[1][0][0], (r[0], r[1][0][1:], r[1][1])))\
    .join(resultadoGanador.map(lambda g: [g[2], g[1]]))\
    .takeSample(False, 6, 25)

[('54832',
  (('982', ['Carlos Jimnez Snchez', '1', '24', '201', '100'], 'ESP'),
   'Silver')),
 ('135219',
  (('944', ['Urka olnir', '2', '22', '173', '64'], 'SLO'), 'Bronze')),
 ('106724',
  (('705', ['Adrie Ard Schenk', '1', '19', '190', '90'], 'NED'), 'Gold')),
 ('112317',
  (('1096', ['Justus Ketchum Smith', '1', '26', '0', '0'], 'USA'), 'Gold')),
 ('124518',
  (('705', ['Daniel Daan van Dijk', '1', '21', '0', '0'], 'NED'), 'Gold')),
 ('42668',
  (('1096', ['Joseph Tilford Lee Joe Greene', '1', '25', '183', '70'], 'USA'),
   'Bronze'))]

Gabriela Barrezueta

student•

No entendí lo de la semilla. con takeSample()

Óscar Gutiérrez Castillo

teacher•

Debido a que takeSample toma una muestra aleatoria, se requiere ingresar una 'semilla' para que se calcule una muestra. Haz la prueba ingresando diferentes semillas.

Frida Ruh

teacher•

Otra forma de entender la semilla es que al ocuparla la computadora escoge valores "semi aleatorios" de forma que si yo ocupo la misma semilla que tú, a ambas nos van a regresar los mismos valores "aleatorios"

Charly Nieves

student•

Mi salida agregando las medallas y el Id del evento a al join ya generado, deportistaOlimpicoRDD.map(lambda l : [l[-1], l[:-1]]).join(equiposOlimpocosRDD.map(lambda x : [x[0], x[2]])).join(resultadoGanador.map( lambda r : [r [-1], r[1]])).take(2)

Cristian Rodríguez

student•

No entendí por qué quería hacer union de los dos csv's. La explicación de que uno de los dos no tiene headers no es suficiente. ¿Supongo que en el otro hay más información necesaria?

Gerardo Mayel Fernández Alamilla

student•

por que al hacer el Join llave-valor la llave se separa del arreglo del resto de valores, por ejemplo ahora si vemos la longitud vemos que se reduce a 1(0,1) como si tuviera únicamente 2 columnas, esto se puede evitar para mantener la misma "estructura" o cómo se puede separar nuevamente ?

Gerardo Sebastian Ramos Reyes

student•

Me costo mucho entender al 100% la sintaxis y armarlo por mi cuenta, bendita IA y sus explicaciones para dumbs

deportistaPaises = deportistaOlimpicosRDD.map(lambda l : [l[-1], l[:-1]]) \ .join(equiposOlimpicosRDD.map(lambda x: [x[0], x[2]])) deportistaPaises.take(1) reto = deportistaPaises.map(lambda l : [l[1][0][0], l[:]]) \ .join(resultadoGanador.map(lambda x : [x[0],x[1:]])) reto.take(1)

Pablo Cristóbal Castro

student•

Agrego mi aporte explicando primero que yo, lo que quería por encima de todo, era trabajar con el id del deportista como clave, parece obvio pero no lo es tanto

teamsAndSportswomenAndScore = teamsAndSportswomen.map(lambda item: [item[1][1][0],item[1]]).join(resultadoGanador.map(lambda item:[item[3],item[:]]))

teamsAndSportswomenAndScore.takeSample(False,20,100)

Pablo Cristóbal Castro

student•

Perdón se me pasó una línea

teamsAndSportswomen = equiposOlimpicosRDD.map(lambda item: [item[0], item[1:]]).join(deportistaOlimpicoRDD.map(lambda item: [item[-1], item[:-1]]))

teamsAndSportswomenAndScore = teamsAndSportswomen.map(lambda item: [item[1][1][0],item[1]]).join(resultadoGanador.map(lambda item:[item[3],item[:]]))

teamsAndSportswomenAndScore.takeSample(False,20,100)

Elias Dudamel

student•

Después de filtrar a solo los deportistas que tuvieron medallas, modifique las posición de las columnas:

resultadosGanadoresOlimpicosRDD = resultadosGanadoresOlimpicosRDD.map(lambda x: (x[2], [element for idx, element in enumerate(x) if idx != 2]))

Y por último hice el join:

deportistaOlimpicoRDD.map(lambda x: (x[-1], x[:-1])) \ .join(equiposOlimpicosRDD.map(lambda x: (x[0], x[-1]))) \ .join(resultadosGanadoresOlimpicosRDD) \ .takeSample(False, 6, 25)

Dando como resultado el siguiente RDD

Solo faltaria sumar el mapeo de cada columna en si mismo y no como objeto anidado.

Francisco Garcia [C6]

student•

Esta mi solución

resultadoGanador.map(lambda l: [l[-1], l[:-1]]) \
.join(deportistaTotalRDD.map(lambda x: [x[0], x[1]])) \
.join(juegosRDD.map(lambda y: [y[0], y[1]])) \
.join(eventoRDD.map(lambda z: [z[2], z[1]])) \
.takeSample(False, 15, 25)

Andres Hita

student•

Hice el join entre lo que ya se había hecho, deportistaOlimpicoRDD y el resultadoGanadorRDD, trayendo todos los campos de resultadoGanadorRDD, usando deportista_id como el punto de la unión.

deportistasYEquipos = deportistaOlimpicoRDD.map(lambda l: [l[-1], l[:-1]]).join(Equipos_Olimpicos_rdd.map(lambda x:[x[0],x[2]]))

deportistasMedalla = deportistasYEquipos.map(lambda x: [x[1][0][0], [x[1][1], x[1][0][1]]]).join(resultadoGanador.map(lambda l: [l[2],[l[0],l[1],l[3],l[4]]]))
deportistasMedalla.takeSample(False, 5, 15)

Output:

[('74148', (['SWE', 'Erik Malmberg'], ['147771', 'Bronze', '9', '280'])),
 ('36026', (['DEN', 'Hans Marius Fogh'], ['70752', 'Silver', '23', '96'])),
 ('18232',
  (['USA', 'Richard John Rick Carey'], ['35396', 'Gold', '35', '150'])),
 ('72276',
  (['SWE', 'Kristina Anna Maria Lundberg'],
   ['144017', 'Silver', '46', '303'])),
 ('46524', (['CAN', 'Gavin Hassett'], ['91794', 'Silver', '41', '399']))]

Alfonso Andres Zapata Guzman

student•

Que tal bro, que bueno no ser el unico haciendo este curso en estas fechas.

~ Que tal Platzinauta, ya conectamos en LinkedIn? ~

¡Que estas esperando! Conectemos en LinkedIn, GitHub, Medium o Redes sociales

Iván Mauricio Jaimes Niño

student•

# Unión de resultadoGanador con equipos y deportistas
deportistaEquipo = deportistaOlimpicoRDD.map(lambda l : [l[-1], l[:-1]]).join(equiposOlimpicosRDD.map(lambda x : [x[0], x[2]]))
deportistaEquipo.take(5)

deportistaEquipo.map(lambda x : [x[1][0][0], x[:]]).join(resultadoGanador.map(lambda y : [y[2], y[:-1]])).take(5)

José Martínez Santana

student•

Reto:

## Union of winners with combined RDD of athetles
winners.map(
    lambda x: [x[2], x[:-1]]
).join(
    athetle_RDD.map(
        lambda y: [y[0], y[:-1]]
    )
).takeSample(False, 6, 25)