PySpark su larga scala

Introduzione a PySpark

Benjamin Schmidt

Data Engineer

Sfruttare la scala

PySpark funziona bene con gigabyte e terabyte di dati
Con PySpark, l'obiettivo è velocità ed efficienza
Capire l'esecuzione di PySpark porta a ulteriori efficienze
Usa broadcast per gestire l'intero cluster

joined_df = large_df.join(broadcast(small_df), 
                          on="key_column", how="inner")
joined_df.show()

Piani di esecuzione

# Usare explain() per vedere il piano di esecuzione
df.filter(df.Age > 40).select("Name").explain()

== Physical Plan ==
*(1) Filter (isnotnull(Age) AND (Age > 30))
+- Scan ExistingRDD[Name:String, Age:Int]

¹ https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.explain.html

Caching e persistenza dei DataFrame

Caching: Memorizza i dati in memoria, per accesso più veloce su dataset piccoli
Persistenza: Memorizza i dati a diversi livelli di storage per dataset grandi

df = spark.read.csv("large_dataset.csv", header=True, inferSchema=True)

# Cache del DataFrame
df.cache()


# Esegui più operazioni sul DataFrame in cache
df.filter(df["column1"] > 50).show()
df.groupBy("column2").count().show()

Persistenza dei DataFrame con diversi livelli di storage

# Persistere il DataFrame con livello di storage
from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)


# Esegui trasformazioni
result = df.groupBy("column3").agg({"column4": "sum"})
result.show()

# Rimuovi la persistenza dopo l'uso
df.unpersist()

Ottimizzare PySpark

Piccole Sottosezioni: Più dati usi, più l'operazione rallenta: Scegli strumenti come map() invece di groupby() per la selettività dei metodi
Join Broadcast: Broadcast utilizza tutta la potenza di calcolo, anche su dataset piccoli
Evita Azioni Ripetute: Azioni ripetute sugli stessi dati costano tempo e risorse, senza vantaggi

Facciamo pratica!

Introduzione a PySpark