Tutorial Avanzado de PySpark y Geopandas para Procesamiento de Datos

BICIMAD y Procesamiento de Datos

BICIMAD: otra línea

Usuarios tipo: df.groupBy("user_type").count().show()

Seleccionar usuarios tipo: sel_type = int(input('Introduce..: '))
df1 = df.filter(df["user_type"] == sel_type)
df1.count() # o sel_type = 2, df1 = df.filter(df.user_type == sele_type)

Uso diario: df_trips = df1.groupBy('user_day_code').count().show()
# df_anual = df.filter(col("user_type") == 1)
viajes_usu = df_anual.groupBy("user_day_code").count()
viajes_usu.filter(col("count") == 1).count()

Estación destino: df.groupBy('idunplug_station').count().show()

Estación destino con distinción: df.groupBy('idunplug_station').agg(count_distinct('idplug_station')).show()

Adyacencia y frecuencia:
edges = df.groupBy("idunplug_station", "idplug_station").count()
result = edges.groupBy("idunplug_station").agg(collect_list(struct(col("count"), col("idplug_station")).alias('destinos')))
result = result.withColumn("destinos_ordenados", sort_array(col("destinos"), asc=False)).show()

Historial de viajes por usuario y día:
hist = df.filter(col("user_type") == 1).groupBy("user_day_code").count().groupBy("count").count().orderBy("count")

Retorno de Viajes y Ciclos

Return trips:
trips = df.groupBy("user_day_code").agg(collect_list(struct("idunplug_station", "idplug_station")).alias("trips")).filter(size(col("trips")) == 2)
def is_return(trips): return int(trips[0][0] == trips[1][1])
udf_return = udf(is_return, IntegerType())
returns = trips.withColumn("return_trip", udf_return(col("trips"))).filter(col("return_trip") == 1)
returns.select(col("trips")[0]["idunplug_station"].alias("origen")).groupBy("origen").count().orderBy(desc("count")).show()

Ciclos:
paths = df.groupBy("user_day_code").agg(collect_list(struct("idunplug_station", "idplug_station")).alias("trips")).filter(size(col("trips")) >= 3)
def is_cycle(trips): start = trips[0][0]; end = trips[-1][1]
return int(start == end)
udf_cycle = udf(is_cycle, IntegerType())
cycles = paths.withColumn("cycle", udf_cycle(col("trips"))).filter(col("cycle") == 1)
cycles.select(col("trips")[0]["idunplug_station"].alias("station")).groupBy("station").count().orderBy(desc("count")).show()

Procesamiento con RDD

Accesos por Día: rdd.filter(lambda x: x[3] == '200').map(lambda x: (x[2], 1)).reduceByKey(lambda a, b: a + b).collect()

Usuarios por Página: rdd.map(lambda x: (x[2], x[0])).groupByKey().mapValues(lambda ips: len(set(ips))).collect()

Mapas y DataFrames en Spark

dataframeSPARK
sdf = spark.read.csv('madrid.csv', header=True, inferSchema=True)
sdf_sample = sdf.filter(sdf.Especie1 == 'Pinus')
gdf = spark_to_geopandas(sdf_sample)
Mapa: gdf.explore(tooltip=['Superficie', '...'])

spark = SparkSession.builder.appName("Ej").getOrCreate()

# Filtrar FormArbol != "No arbolado"
df_arbol = sdf.filter(col("FormArbol") != "No arbolado")

gdf_arbolado = spark_to_geopandas(df_arbolado)

gdf_arbolado.explore()

# Proporción de Superficie < 1 hectárea y >= 1 hectárea
total = sdf.count()
small = sdf.filter(col("Superficie") < 1).count()

large = sdf.filter(col("Superficie") >= 1).count()

small / total, large / total

# Número de registros por cada valor posible
form_count = sdf.groupBy("FormArbol").count().orderBy(col("count").desc())
form_count.show()

Agregaciones y Estadísticas Espaciales

Suma de las superficies de cada una de las categorías de FormArbol. ¿Cuáles son las 5 con mayor extensión?

surf_total = sdf.groupBy("FormArbol").sum("Superficie").withColumnRenamed("sum(Superficie)", "superficie_total")
surf_total.orderBy(col("superficie_total").desc()).show(5)

Superficie total, media y número de parcelas:
sdf.groupBy("FormArbol").agg(sum("Superficie").alias("sup_total"), avg("Sup").alias("sup_media"), count("*").alias("num_parce")).show()

¿Categoría FormArbol solo asociada a un tipo de Destino?
rel = sdf.groupBy("FormArbol").agg(countDistinct("DesTipEstr").alias("n_tipos"))
rel.filter(col("n_tipos") == 1).show()
# Valor asociado a DesTipEstr en ese caso:
sdf.groupBy("FormArbol") \ .agg(collect_set("DesTipEstr").alias("tipos")) \ .filter(size(col("tipos")) == 1) \ .show(truncate=False)

Recopilar conjunto de especies que aparecen con cada una de las especies principales:
sdf.groupBy("Especie1").agg(collect_set("Especie2").alias("Especies_secundarias"), collect_set("Especie3").alias("Especies_terciarias")).show(truncate=False)

Otros Análisis Adicionales

Parcelas más grandes por especie:
sdf.groupBy("Especie1").max("Superficie").show()

FormArbol dominante por especie:
from pyspark.sql.functions import desc
sdf.groupBy("Especie1", "FormArbol").count() \ .orderBy(desc("count")).show()

Parcelas sin especie secundaria:
sdf.filter(col("Especie2").isNull()).count()

Distribución espacial de una especie:
spark_to_geopandas(sdf.filter(col("Especie1") == "Pinus sylvestris")).explore()