Apuntes del curso en Notion

Importando Spark

from pyspark import SparkContext from pyspark.sql import SparkSession

Creando la Session

Una vez creada la session podemos realizar todas las operaciones que spark permite como por ejemplo crear RDD, Dataframe, importar data externa, etc.

spark = SparkSession.builder \ .master("local") \ .appName("FirstSession") \ .getOrCreate()

spark

Terminar la session o context para que no siga en memoria consumiendo recursos del Cluster con .stop()

spark.stop()

Creando Context

SparkContext Permite crear un session para comenzar a utilizar spark

sc = SparkContext("local", "First App")

Podemos pasar Context a Session de la siguiente manera

spark2 = SparkSession(sc)

spark2

RDD

Transformaciones y acciones

rdd1=sc.parallelize([1,2,3]) type(rdd1)

Collect

Podemos verificar la data del RDD con la siguiente acción

#No utiizar collect() en big data puede congestionar la base , lo mejor es take(10) rdd1.collect()

Podemos cargar csv en rdd. Recordar que si tenemos algún error solo sera posible ver cuando se ejecute un action

equipos_rdd = sc.textFile('files/paises.csv')\ .map(lambda line: line.split(',') )

Take

.take() para ver los registros del rdd

equipos_rdd.take(15)

Exploramos un Rdd con .map() que recibe una función lambda por ejemplo y aplica dicha función a cada elemento en el rdd y devuelve un nuevo rdd, vamos a ver un ejemplo en la siguiente celda con un conteo de los distintos elemento de la columna 'sigla'. [ x[2] ] -> Procesando como una lista. ( x[2] ) ->Procesando como una tupla

equipos_rdd.map(lambda x: ( x[2] ) ).take(5)

equipos_rdd.map(lambda x: ( x[2] ) ).distinct().count()

GroupBy

Vamos a utilizar .groupBy() para agrupar la data y .map() para seleccionar primero la columna key por la que necesitamos la agrupación y segundo la columna que vamos a realizar el count. .mapValues(len or list or xfunction) -> retorna la data en base a lo que recibe como parámetro

equipos_rdd.map(lambda x: ( x[2],x[1] )).take(5)

equipos_rdd.map(lambda x: ( x[2],x[1] )).groupByKey().mapValues(len).take(20)

Filter

Podemos realizar filtros de data en rdd con .filter()

#Ojo con el collect solo con data pequeña, #puede ocasionar lentitud en servidor si se quiere ver millones equipos_rdd.filter(lambda pais: "BRA" in pais).collect()

Count

Si crees que tienes un rdd inmenso puedes realizar un .countApprox(20) y enviarle como param el tiempo en milisegundo para que solo cuente hasta ese momento

equipos_rdd.countApprox(60)

Union

Vamos a realizar uniones de Rdd con el .union()

deportistas_rdd1=sc.textFile('files/deportista.csv').map(lambda l: l.split(',')) deportistas_rdd1.take(5)

deportistas_rdd2=sc.textFile('files/deportista2.csv').map(lambda l: l.split(',')) deportistas_rdd2.take(5)

deportistas_rdd = deportistas_rdd1.union(deportistas_rdd2) deportistas_rdd.take(5)

deportistas_rdd.countApprox(60)

Join

Podemos utilizar .join() para realizar cruces en rdd. .top() -> Nos permite observar las dos primeras registros del dataset

equipos_rdd.top(2)

deportistas_rdd.top(2)

Vamos a ordenar las columnas del rdd de tal manera que quede al inicio el campo por el cual se va a hacer join

deportistas_rdd.map(lambda pk: [pk[-1],pk[:-1]] ).take(2)

equipos_rdd.map(lambda fk: [fk[0],fk[2]] ).take(2)

Uniremos mediante join las dos celdas anteriores y utilizamos .takeSample() para elegir una muestra aleatorias del rdd y se le envía como parámetros True o False para repetidos, la cantidad de registros que quiero ver y por ultimo un numero aleatorio o semilla que va a mostrar los mismos registros siempre

deportistas_equipos_rdd=deportistas_rdd.map(lambda pk: [pk[-1],pk[:-1]] )\ .join(equipos_rdd.map(lambda fk: [fk[0],fk[2]] )) deportistas_equipos_rdd.takeSample(False,6,25)

Creamos un nuevo rdd para los ganadores de medallas olimpicas

resultados_rdd = sc.textFile('files/resultados.csv') \ .map(lambda l: l.split(','))

resultados_rdd.take(4)

Vamos a realizar una agrupación y conteo del campo medalla para ver los distintos valores que tiene

resultados_rdd.map(lambda data: data[1] ).distinct().count()

Como vemos que son pocos registros tranquilamente podemos usar un .collect() para agrupar

resultados_rdd.map(lambda data: (data[1],data[1]) ).groupByKey().mapValues(len).collect()

Filtramos la data no queremos los NA

ganadores_rdd= resultados_rdd.filter(lambda data: 'NA' not in data[1]) ganadores_rdd.map(lambda data: (data[1],data[1]) ).groupByKey().mapValues(len).collect()

Reto: Haremos un join del deportistas_equipos_rdd que contiene los equipos y los deportista que pertenecen a cada uno de ellos con los ganadores_rdd que tienen solo los que ganaron medallas

deportistas_equipos_rdd.top(2)

Necesitamos solo el campo deportista_id y las siglas del pais al que pertenecen para esto utilizamos slices

deportistas_equipos_rdd.map(lambda pk: (pk[1][0][0], pk[1][1] ) ).takeSample(False,5,25)

ganadores_rdd.top(2)

Hacemos el join el primer dato es el campo 'deportista_id' de ambos rdd

deportistas_equipos_ganadores_rdd=deportistas_equipos_rdd.map(lambda pk: (pk[1][0][0], pk[1][1] ) )\ .join(ganadores_rdd.map(lambda fk: (fk[2],fk[1]) )) deportistas_equipos_ganadores_rdd.takeSample(False,5,25)

Cuando hacemos join se crean sub tuplas y para hacer slices para acceder a esos valores es de la siguiente manera

deportistas_equipos_ganadores_rdd.map(lambda data: ( data[1][0] ,data[1][1] ) ).takeSample(False,5,25)

add

from operator import add

Vamos a crear un diccionarios con el puntaje que le asignan a cada pais por medalla

puntos_dict= { 'Gold': 7, 'Bronze': 5, 'Silver': 4} puntos_dict

Utilizamos el map del rdd donde tenemos el pais y la medalla ganada y Aplicamos el diccionario donde tenemos mapeados los puntajes de las medallas

puntajes_paises_rdd= \ deportistas_equipos_ganadores_rdd.map(lambda data: ( data[1][0] ,puntos_dict[data[1][1]] ) ) puntajes_paises_rdd.takeSample(False,5,25)

Por ultimo aplicamos una sumatoria por paises y ordenamos para obtener el top 5 de los mas ganadores

puntajes_paises_rdd.reduceByKey((add)) \ .sortBy(lambda data: data[1], ascending=False).take(5)

Dataframe

from pyspark.sql.types import StructType, StructField, IntegerType,\ StringType, FloatType, Row from pyspark.sql import *

sql_context = SQLContext(sc)

!head -n 5 'files/juegos.csv'

Load

Vamos a crear el esquema para el dataframe

juegos_schema = StructType( [ StructField('juego_id',IntegerType(),False), StructField('anio',StringType(),False), StructField('Temporada',StringType(),False), StructField('Ciudad',StringType(),False) ] )

juegos_df = sql_context.read.schema(juegos_schema).option('header','true').csv('files/juegos.csv') juegos_df.show(5)

Otra manera de cargar dataframe en las siguientes celdas

equipos_df = spark2.read.option('header','true').csv('files/paises.csv') equipos_df.show(5)

deportistas_df1 = spark2.read.option('header','true').csv('files/deportista.csv') deportistas_df1.show(2)

deportistas_df2 = spark2.read.csv('files/deportista2.csv') deportistas_df2.show(2)

deportistas_df = deportistas_df1.union(deportistas_df2) deportistas_df.show(5)

resultados_df = spark2.read.option('header','true').csv('files/resultados.csv') resultados_df.show(5)

eventos_df = spark2.read.option('header','true').csv('files/evento.csv') eventos_df.show(5)

Operaciones con df

deportistas_df.printSchema()

En la siguiente celdas podemos ver como renombrar y eliminar columnas del df

deportistas_df = deportistas_df.withColumnRenamed('genero','sexo').drop('altura') deportistas_df.printSchema()

Filter

Filtrando y ordenando en df

deportistas_df.filter( deportistas_df.edad != 0).sort('edad',ascending=False).show(5)

Select

Select en df y para utilizar col().alias() debemos importar la siguiente librería

from pyspark.sql.functions import *

deportistas_final_df = deportistas_df.select('deportista_id','nombre','sexo',col('edad').alias('edad_al_jugar'),'peso','equipo_id')

deportistas_final_df.printSchema()

Joins

deportistas_final_df.printSchema()

resultados_df.printSchema()

juegos_df.printSchema()

eventos_df.printSchema()

equipos_df.printSchema()

Teniendo definidos los esquemas podemos empezar a realizar joins

deportistas_final_df\ .join(equipos_df, equipos_df.id == deportistas_final_df.equipo_id,"left")\ .join(resultados_df, resultados_df.deportista_id == deportistas_final_df.deportista_id,"left")\ .join(juegos_df, juegos_df.juego_id == resultados_df.juego_id, "left")\ .join(eventos_df, eventos_df.evento_id == resultados_df.evento_id, "left")\ .select(deportistas_final_df.nombre,deportistas_final_df.edad_al_jugar,equipos_df.sigla \ ,resultados_df.medalla, juegos_df.anio, eventos_df.evento )\ .sort(juegos_df.anio,ascending=False)\ .show(6)

Reto: Hacer Joins de las medallas y los equipos que las ganaron, no mostrar los equipos que no ganaron medallas

ganadores_df = resultados_df.filter(resultados_df.medalla != 'NA') ganadores_df.show()

Tenemos que realizar un join de los deportistas con los equipos o selecciones a la que pertenecen y por ultimo otro join con el df de solo ganadores

pais_ganadores_df=deportistas_final_df\ .join(equipos_df, equipos_df.id == deportistas_final_df.equipo_id,"left")\ .join(ganadores_df,ganadores_df.deportista_id == deportistas_final_df.deportista_id,"inner")\ .select("sigla", "medalla") pais_ganadores_df.show(5)

GroupBy

total_medallas_df=pais_ganadores_df.groupBy('sigla','medalla').count().sort('count',ascending=False) total_medallas_df.printSchema()

total_medallas_df.groupBy('sigla').agg(sum('count').alias('Total medallas'))\ .sort('Total medallas',ascending=False).show()

Sql

Convertimos los df en tablas temporales dentro de spark pero el verdadero poder de spark se ve en usar las funciones nativas de los df

resultados_df.createOrReplaceTempView('resultados') deportistas_final_df.createOrReplaceTempView('deportistas') equipos_df.createOrReplaceTempView('equipos')

sql_context.sql("SELECT * FROM resultados").show(1)

sql_context.sql("SELECT * FROM deportistas").show(1)

sql_context.sql("SELECT * FROM equipos").show(1)

sql_context.sql(""" SELECT e.sigla, r.medalla, count(1) FROM resultados r JOIN deportistas d ON r.deportista_id=d.deportista_id JOIN equipos e ON e.id=d.equipo_id WHERE r.medalla <> 'NA' GROUP BY e.sigla, r.medalla ORDER BY count(1) desc """).show()

Udf

UDF (User Defined Functions) son las funciones de usuario, y son sistemas para definir nuevos métodos SQL que operan sobre las columnas de un DataFrame

from pyspark.sql.functions import udf

deportistas_error_df = spark2.read.option('header','true').csv('files/deportistaError.csv') deportistas_error_df.show(20)

Creamos una función normal

def convertir_enteros(valor): return int(valor) if valor != None else 0

Registramos la udf en spark con las siguientes líneas

convertir_enteros_udf = udf(lambda z: convertir_enteros(z), IntegerType()) sql_context.udf.register("convertir_enteros_udf",convertir_enteros_udf)

deportistas_error_df.select(convertir_enteros_udf("altura")).show()

Persistencia

from pyspark.storagelevel import StorageLevel

Vamos a validar si un df esta almacenado en cache

equipos_df.is_cached

Para mantener el df en cache podemos ejecutar la siguiente propiedad

equipos_df.rdd.cache()

Para saber cual es el tipo de persistencia que tiene ejecutamos la siguiente sentencia y podemos validar en el link que significa su salida

equipos_df.rdd.getStorageLevel()

Vamos a bajar de la persistencia en cache el df y aplicar una persistencia en disco y cache memoria

equipos_df.rdd.unpersist() equipos_df.rdd.persist(StorageLevel.MEMORY_AND_DISK_2) equipos_df.rdd.getStorageLevel()

Vamos agregar un numero mas de replicación al df

StorageLevel.MEMORY_AND_DISK_3 = StorageLevel(True,True,False,False,3)

equipos_df.rdd.unpersist() equipos_df.rdd.persist(StorageLevel.MEMORY_AND_DISK_3) equipos_df.rdd.getStorageLevel()

Particionado

from pyspark.sql import SparkSession spark_part = SparkSession.builder.appName("Particionado").master("local[5]")\ .getOrCreate()

df = spark_part.range(0,20) df.rdd.getNumPartitions()

.css-15w88e5{color:var(--chakra-colors-fg-neutral-primary);font-weight:inherit;letter-spacing:-0.09px;}Apuntes del curso en Notion

Importando Spark

Creando la Session

Creando Context

RDD

Collect

Take

GroupBy

Filter

Count

Union

Join

add

Dataframe

Load

Operaciones con df

Filter

Select

Joins

GroupBy

Sql

Udf

Persistencia

Particionado

Apuntes del curso en Notion