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)

/root/venv/lib/python3.7/site-packages/pyspark/sql/context.py:79: FutureWarning: Deprecated in 3.0.0. Use SparkSession.builder.getOrCreate() instead.
  FutureWarning

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

,nombre_juego,annio,temporada,ciudad
1,1896 Verano,1896,Verano,Athina
2,1900 Verano,1900,Verano,Paris
3,1904 Verano,1904,Verano,St. Louis
4,1906 Verano,1906,Verano,Athina

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)

+--------+-----------+---------+------+
|juego_id|       anio|Temporada|Ciudad|
+--------+-----------+---------+------+
|       1|1896 Verano|     1896|Verano|
|       2|1900 Verano|     1900|Verano|
|       3|1904 Verano|     1904|Verano|
|       4|1906 Verano|     1906|Verano|
|       5|1908 Verano|     1908|Verano|
+--------+-----------+---------+------+
only showing top 5 rows

Otra manera de cargar dataframe en las siguientes celdas

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

+---+--------------------+-----+
| id|              equipo|sigla|
+---+--------------------+-----+
|  1|         30. Februar|  AUT|
|  2|A North American ...|  MEX|
|  3|           Acipactli|  MEX|
|  4|             Acturus|  ARG|
|  5|         Afghanistan|  AFG|
+---+--------------------+-----+
only showing top 5 rows

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

+-------------+---------+------+----+------+----+---------+
|deportista_id|   nombre|genero|edad|altura|peso|equipo_id|
+-------------+---------+------+----+------+----+---------+
|            1|A Dijiang|     1|  24|   180|  80|      199|
|            2| A Lamusi|     1|  23|   170|  60|      199|
+-------------+---------+------+----+------+----+---------+
only showing top 2 rows

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

+-----+----------+---+---+---+---+---+
|  _c0|       _c1|_c2|_c3|_c4|_c5|_c6|
+-----+----------+---+---+---+---+---+
|67787|Lee BongJu|  1| 27|167| 56|970|
|67788|  Lee BuTi|  1| 23|164| 54|203|
+-----+----------+---+---+---+---+---+
only showing top 2 rows

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

+-------------+--------------------+------+----+------+----+---------+
|deportista_id|              nombre|genero|edad|altura|peso|equipo_id|
+-------------+--------------------+------+----+------+----+---------+
|            1|           A Dijiang|     1|  24|   180|  80|      199|
|            2|            A Lamusi|     1|  23|   170|  60|      199|
|            3| Gunnar Nielsen Aaby|     1|  24|     0|   0|      273|
|            4|Edgar Lindenau Aabye|     1|  34|     0|   0|      278|
|            5|Christine Jacoba ...|     2|  21|   185|  82|      705|
+-------------+--------------------+------+----+------+----+---------+
only showing top 5 rows

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

+------------+-------+-------------+--------+---------+
|resultado_id|medalla|deportista_id|juego_id|evento_id|
+------------+-------+-------------+--------+---------+
|           1|     NA|            1|      39|        1|
|           2|     NA|            2|      49|        2|
|           3|     NA|            3|       7|        3|
|           4|   Gold|            4|       2|        4|
|           5|     NA|            5|      36|        5|
+------------+-------+-------------+--------+---------+
only showing top 5 rows

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

+---------+--------------------+----------+
|evento_id|              evento|deporte_id|
+---------+--------------------+----------+
|        1|Basketball Men's ...|         1|
|        2|Judo Men's Extra-...|         2|
|        3|Football Men's Fo...|         3|
|        4|Tug-Of-War Men's ...|         4|
|        5|Speed Skating Wom...|         5|
+---------+--------------------+----------+
only showing top 5 rows

Operaciones con df

deportistas_df.printSchema()

root
 |-- deportista_id: string (nullable = true)
 |-- nombre: string (nullable = true)
 |-- genero: string (nullable = true)
 |-- edad: string (nullable = true)
 |-- altura: string (nullable = true)
 |-- peso: string (nullable = true)
 |-- equipo_id: string (nullable = true)

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

root
 |-- deportista_id: string (nullable = true)
 |-- nombre: string (nullable = true)
 |-- sexo: string (nullable = true)
 |-- edad: string (nullable = true)
 |-- peso: string (nullable = true)
 |-- equipo_id: string (nullable = true)

Filter

Filtrando y ordenando en df

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

+-------------+--------------------+----+----+----+---------+
|deportista_id|              nombre|sexo|edad|peso|equipo_id|
+-------------+--------------------+----+----+----+---------+
|       128719|John Quincy Adams...|   1|  97|   0|     1096|
|        49663|       Winslow Homer|   1|  96|   0|     1096|
|        31173|Thomas Cowperthwa...|   1|  88|   0|     1096|
|       118789|        Louis Tauzin|   1|  81|   0|      362|
|        69729|      Max Liebermann|   1|  80|   0|      399|
+-------------+--------------------+----+----+----+---------+
only showing top 5 rows

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

root
 |-- deportista_id: string (nullable = true)
 |-- nombre: string (nullable = true)
 |-- sexo: string (nullable = true)
 |-- edad_al_jugar: string (nullable = true)
 |-- peso: string (nullable = true)
 |-- equipo_id: string (nullable = true)

Joins

deportistas_final_df.printSchema()

root
 |-- deportista_id: string (nullable = true)
 |-- nombre: string (nullable = true)
 |-- sexo: string (nullable = true)
 |-- edad_al_jugar: string (nullable = true)
 |-- peso: string (nullable = true)
 |-- equipo_id: string (nullable = true)

resultados_df.printSchema()

root
 |-- resultado_id: string (nullable = true)
 |-- medalla: string (nullable = true)
 |-- deportista_id: string (nullable = true)
 |-- juego_id: string (nullable = true)
 |-- evento_id: string (nullable = true)

juegos_df.printSchema()

root
 |-- juego_id: integer (nullable = true)
 |-- anio: string (nullable = true)
 |-- Temporada: string (nullable = true)
 |-- Ciudad: string (nullable = true)

eventos_df.printSchema()

root
 |-- evento_id: string (nullable = true)
 |-- evento: string (nullable = true)
 |-- deporte_id: string (nullable = true)

equipos_df.printSchema()

root
 |-- id: string (nullable = true)
 |-- equipo: string (nullable = true)
 |-- sigla: string (nullable = true)

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)

+-----------------+-------------+-----+-------+-----------+--------------------+
|           nombre|edad_al_jugar|sigla|medalla|       anio|              evento|
+-----------------+-------------+-----+-------+-----------+--------------------+
|Nstor Abad Sanjun|           23|  ESP|     NA|2016 Verano|Gymnastics Men's ...|
|Giovanni Abagnale|           21|  ITA| Bronze|2016 Verano|Rowing Men's Coxl...|
|Nstor Abad Sanjun|           23|  ESP|     NA|2016 Verano|Gymnastics Men's ...|
|Nstor Abad Sanjun|           23|  ESP|     NA|2016 Verano|Gymnastics Men's ...|
|Nstor Abad Sanjun|           23|  ESP|     NA|2016 Verano|Gymnastics Men's ...|
|Nstor Abad Sanjun|           23|  ESP|     NA|2016 Verano|Gymnastics Men's ...|
+-----------------+-------------+-----+-------+-----------+--------------------+
only showing top 6 rows

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

+------------+-------+-------------+--------+---------+
|resultado_id|medalla|deportista_id|juego_id|evento_id|
+------------+-------+-------------+--------+---------+
|           4|   Gold|            4|       2|        4|
|          38| Bronze|           15|       7|       19|
|          39| Bronze|           15|       7|       20|
|          41| Bronze|           16|      50|       14|
|          42| Bronze|           17|      17|       21|
|          43|   Gold|           17|      17|       22|
|          45|   Gold|           17|      17|       24|
|          49|   Gold|           17|      17|       28|
|          51| Bronze|           17|      19|       22|
|          61|   Gold|           20|      38|       32|
|          62| Bronze|           20|      38|       33|
|          64| Silver|           20|      40|       31|
|          65| Bronze|           20|      40|       32|
|          68| Silver|           20|      40|       35|
|          74|   Gold|           20|      44|       32|
|          77|   Gold|           20|      44|       35|
|          79|   Gold|           20|      46|       32|
|          80|   Gold|           21|      47|       36|
|          87| Silver|           25|       7|       41|
|          92| Bronze|           29|      37|       45|
+------------+-------+-------------+--------+---------+
only showing top 20 rows

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)

+-----+-------+
|sigla|medalla|
+-----+-------+
|  SWE|   Gold|
|  FIN| Bronze|
|  FIN| Bronze|
|  FIN| Bronze|
|  FIN| Bronze|
+-----+-------+
only showing top 5 rows

GroupBy

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

root
 |-- sigla: string (nullable = true)
 |-- medalla: string (nullable = true)
 |-- count: long (nullable = false)

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

+-----+--------------+
|sigla|Total medallas|
+-----+--------------+
|  USA|          5643|
|  URS|          2660|
|  GBR|          2045|
|  GER|          2045|
|  FRA|          1785|
|  ITA|          1627|
|  SWE|          1537|
|  CAN|          1347|
|  AUS|          1314|
|  HUN|          1131|
|  GDR|          1084|
|  NED|          1039|
|  NOR|          1033|
|  RUS|          1003|
|  CHN|           991|
|  JPN|           913|
|  FIN|           900|
|  SUI|           690|
|  ROU|           658|
|  KOR|           642|
+-----+--------------+
only showing top 20 rows

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)

+------------+-------+-------------+--------+---------+
|resultado_id|medalla|deportista_id|juego_id|evento_id|
+------------+-------+-------------+--------+---------+
|           1|     NA|            1|      39|        1|
+------------+-------+-------------+--------+---------+
only showing top 1 row

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

+-------------+---------+----+-------------+----+---------+
|deportista_id|   nombre|sexo|edad_al_jugar|peso|equipo_id|
+-------------+---------+----+-------------+----+---------+
|            1|A Dijiang|   1|           24|  80|      199|
+-------------+---------+----+-------------+----+---------+
only showing top 1 row

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

+---+-----------+-----+
| id|     equipo|sigla|
+---+-----------+-----+
|  1|30. Februar|  AUT|
+---+-----------+-----+
only showing top 1 row

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

+-----+-------+--------+
|sigla|medalla|count(1)|
+-----+-------+--------+
|  USA|   Gold|    2639|
|  USA| Silver|    1648|
|  USA| Bronze|    1356|
|  URS|   Gold|    1137|
|  URS| Silver|     783|
|  URS| Bronze|     740|
|  GBR| Silver|     720|
|  GER| Bronze|     709|
|  GER|   Gold|     690|
|  GBR|   Gold|     675|
|  FRA| Bronze|     668|
|  GBR| Bronze|     650|
|  GER| Silver|     646|
|  FRA| Silver|     613|
|  ITA|   Gold|     574|
|  SWE| Bronze|     535|
|  ITA| Bronze|     528|
|  ITA| Silver|     525|
|  SWE| Silver|     522|
|  AUS| Bronze|     511|
+-----+-------+--------+
only showing top 20 rows

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)

+-------------+--------------------+------+----+------+----+---------+
|deportista_id|              nombre|genero|edad|altura|peso|equipo_id|
+-------------+--------------------+------+----+------+----+---------+
|            1|           A Dijiang|     1|  24|   180|  80|      199|
|            2|            A Lamusi|     1|  23|   170|  60|      199|
|            3| Gunnar Nielsen Aaby|     1|  24|  null|null|      273|
|            4|Edgar Lindenau Aabye|     1|  34|  null|null|      278|
|            5|Christine Jacoba ...|     2|  21|   185|  82|      705|
|            6|     Per Knut Aaland|     1|  31|   188|  75|     1096|
|            7|        John Aalberg|     1|  31|   183|  72|     1096|
|            8|"Cornelia ""Cor""...|     2|  18|   168|null|      705|
|            9|    Antti Sami Aalto|     1|  26|   186|  96|      350|
|           10|"Einar Ferdinand ...|     1|  26|  null|null|      350|
|           11|  Jorma Ilmari Aalto|     1|  22|   182|76.5|      350|
|           12|   Jyri Tapani Aalto|     1|  31|   172|  70|      350|
|           13|  Minna Maarit Aalto|     2|  30|   159|55.5|      350|
|           14|Pirjo Hannele Aal...|     2|  32|   171|  65|      350|
|           15|Arvo Ossian Aaltonen|     1|  22|  null|null|      350|
|           16|Juhamatti Tapio A...|     1|  28|   184|  85|      350|
|           17|Paavo Johannes Aa...|     1|  28|   175|  64|      350|
|           18|Timo Antero Aaltonen|     1|  31|   189| 130|      350|
|           19|Win Valdemar Aalt...|     1|  54|  null|null|      350|
|           20|  Kjetil Andr Aamodt|     1|  20|   176|  85|      742|
+-------------+--------------------+------+----+------+----+---------+
only showing top 20 rows

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

+----------------+
|<lambda>(altura)|
+----------------+
|             180|
|             170|
|               0|
|               0|
|             185|
|             188|
|             183|
|             168|
|             186|
|               0|
|             182|
|             172|
|             159|
|             171|
|               0|
|             184|
|             175|
|             189|
|               0|
|             176|
+----------------+
only showing top 20 rows

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