Aggregazioni PySpark

Introduzione a PySpark

Benjamin Schmidt

Data Engineer

Panoramica sulle aggregazioni SQL in PySpark

Le comuni aggregazioni SQL funzionano con spark.sql()

# Query di aggregazione SQL
spark.sql("""
  SELECT Department, SUM(Salary) AS Total_Salary, AVG(Salary) AS Average_Salary
  FROM employees
  GROUP BY Department
""").show()

Combinare operazioni DataFrame e SQL

# Filtra stipendi sopra i 3000
filtered_df = df.filter(df.Salary > 3000)

# Registra il DataFrame filtrato come vista
filtered_df.createOrReplaceTempView("filtered_employees")


# Aggrega usando SQL sulla vista filtrata
spark.sql("""
    SELECT Department, COUNT(*) AS Employee_Count
    FROM filtered_employees
    GROUP BY Department
""").show()

Gestione dei tipi di dati nelle aggregazioni

# Esempio di conversione di tipo
data = [("HR", "3000"), ("IT", "4000"), ("Finance", "3500")]
columns = ["Department", "Salary"]
df = spark.createDataFrame(data, schema=columns)


# Converti la colonna Salary in intero
df = df.withColumn("Salary", df["Salary"].cast("int"))

# Esegui l'aggregazione
df.groupBy("Department").sum("Salary").show()

RDD per le aggregazioni

# Esempio di aggregazione con RDD
rdd = df.rdd.map(lambda row: (row["Department"], row["Salary"]))


rdd_aggregated = rdd.reduceByKey(lambda x, y: x + y)


print(rdd_aggregated.collect())

Best practice per le aggregazioni PySpark

Filtra presto: Riduci la dimensione dei dati prima delle aggregazioni
Gestisci i tipi di dati: Assicurati che i dati siano puliti e correttamente tipizzati
Evita operazioni sull'intero dataset: Minimizza operazioni come groupBy()
Scegli l'interfaccia giusta: Preferisci i DataFrame per la maggior parte dei compiti grazie alle loro ottimizzazioni
Monitora le prestazioni: Usa explain() per controllare il piano di esecuzione e ottimizzare di conseguenza

Punti chiave

Aggregazioni SQL in PySpark: Funzioni come SUM() e AVERAGE() per riassumere i dati
DataFrames e SQL: Combinare entrambi per una manipolazione dati flessibile
Gestione dei tipi di dati: Risolvere problemi di incompatibilità durante le aggregazioni
RDD vs DataFrames: Capire i compromessi e scegliere lo strumento giusto

Mettiamoci alla prova!

Introduzione a PySpark