Resolución de Ejercicios Prácticos con PySpark y SQL para Procesamiento de Datos

Resolución de Ejercicios Prácticos con PySpark

A continuación, se presenta la solución estructurada para una serie de ejercicios prácticos utilizando PySpark y Spark SQL, enfocados en el procesamiento y transformación de datos electorales y municipales.

Ejercicio 13

Se registra la vista temporal de municipios y se procesa el dataset de elecciones combinándolo mediante SQL:

df_municipios.createOrReplaceTempView('municipios') # lo mismo con elecciones

df13 = spark.sql("""select b.Comunidad, b.Provincia, b.Municipio, stack(51, lo de antes) as (partido, votos), a.Votantes from elecciones a join municipios b on b.Codigo = a.Codigo""") 
df13.createOrReplaceTempView('df13')

Posteriormente, se realizan agrupaciones y cálculos estadísticos avanzados utilizando funciones de ventana (window functions) filtrando por provincia y partidos específicos:

df13_2 = spark.sql("""select distinct Comunidad, Provincia, partido, 
round(avg(votos) over (partition by Provincia, partido),2) as Media_votos,
sum(votos) over (partition by Provincia, partido) as Votos_total,
min(votos) over (partition by Provincia, partido) as Min_votos,
max(votos) over (partition by Provincia, partido) as Max_votos,
count(Municipio) over (partition by Provincia, partido) as Num_municipios,
round(sum(votos) over (partition by Comunidad, Provincia, partido) / sum(Votantes) over (partition by Comunidad, Provincia, partido) * 100, 2) as Percentage
from df13 where Provincia = 'Murcia' and (partido = 'EB' or partido = 'PSOE')""")
df13_2.createOrReplaceTempView('df13_2')

Ejercicio 5

Cruzamiento de DataFrames y reestructuración de columnas mediante la función stack:

df5 = df_elecciones.join(df_municipios, on = 'Codigo')
df5_stacked = df5.select('Municipio', expr("stack(51, lo del pdf) as (partido, votos)"))

Ejercicio 6

Extracción de los registros con menor población (sección de ordenamiento ascendente y guardado en formato CSV):

df_top = df_municipios.sort(df_municipios['Poblacion'], ascending=True).select('Comunidad', 'Provincia', 'Municipio', 'Poblacion').limit(10)
df_top.write.format('csv').option("delimiter", ";").save('<a>hdfs://localhost:9000/Examen_practico/Resultados/Ejer_6_top</a>') # ls y cat

Ejercicio 7

Creación de rangos de participación electoral y cálculo de abstenciones mediante condiciones lógicas:

df_elecciones1 = df_elecciones.withColumn("rango_participacion", when(((col('Votantes')/col('Censo'))*100) == 100, "TOTAL")\
    .when(((col('Votantes')/col('Censo'))*100) > 65, "ALTA")\
    .when((((col('Votantes')/col('Censo'))*100) <= 65) & (((col('Votantes')/col('Censo'))*100) > 30), "PARCIAL")\
    .otherwise('BAJA'))

df_elecciones2 = df_elecciones1.withColumn("participacion", round(col('Votantes')/col('Censo')*100, 2)).withColumn('abstencion', round(100-col('participacion'),2)).select('Codigo', 'participacion','abstencion', 'rango_participacion')
df7 = df_elecciones2.join(df_municipios, on='Codigo').sort('Codigo')
df7_top = df7.sort('participacion', ascending=False).limit(15) # lo mismo con bottom y hacer el write y ls y cat

Ejercicio 12

Procesamiento avanzado utilizando la API de RDDs para sumar votos y encontrar municipios destacados por comunidad:

df_ej5 = df_e.select('Codigo_E', expr("stack(51, lo de siempre) as (partidos, votos)"))
df_ej5_all = df_ej5.join(df_m, df_m.Codigo_M == df_ej5.Codigo_E)
df_ej5_rdd = df_ej5_all.rdd

sumatorio_comunidad = df_ej5_rdd.map(lambda x: (x.Comunidad, x.votos)).reduceByKey(lambda x, y: x+y)
municipios_mas_votados = df_ej5_rdd.map(lambda x: (x.Comunidad, (x.Municipio, x.votos))).reduceByKey(lambda x, y: max(x, y, key = lambda maximo: maximo[-1]))

join_valor_porcentual = municipios_mas_votados.join(sumatorio_comunidad)
rdd_ej12 = join_valor_porcentual.map(lambda x: (x[0], x[1][0][0], round((x[1][0][1]/x[1][1])*100, 2) ))
rdd_ej12.saveAsTextFile("<a style=\"background-color:#ffffff\">hdfs://localhost:9000/Examen_practico/Resultados/Ejer_12_top</a>.txt")

Ejercicio 11

Agrupación por provincia, cálculo de porcentajes y aplicación de particiones con ventanas para obtener el ranking de partidos principales:

df11_stacked = df5.select('Provincia', expr("stack(51, lo de siempre) as (partido, votos)"))
df11_stacked_grouped = df11_stacked.groupBy('Provincia').agg(sum('votos').alias('votos_totales'))

df11 = df_prob.join(df11_stacked, on='Provincia', how='left').join(df11_stacked_grouped.select('Provincia', 'votos_totales'), on='Provincia', how='left').withColumn('porcentaje_votos', round(col('votos')/col('votos_totales')*100, 2))
df11.groupBy('Comunidad','Provincia').agg(sum('porcentaje_votos')).show() # df11.printSchema()

window = Window.partitionBy(col('Provincia'))\
    .orderBy(col('porcentaje_votos').desc())
df11_partidos = df11.select('Comunidad', 'Provincia','partido', 'porcentaje_votos', rank().over(window).alias('rank')).filter(col('rank') <= 3)