PySpark à grande échelle

Introduction à PySpark

Benjamin Schmidt

Data Engineer

Tirer parti de l'échelle

PySpark gère efficacement des giga‑ et téraoctets de données
Avec PySpark, l'objectif est la vitesse et un traitement efficace
Comprendre l'exécution PySpark accroît encore l'efficacité
Utilisez le « broadcast » pour gérer tout le cluster

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

Plans d'exécution

# Utiliser explain() pour voir le plan d'exécution
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

Mettre en cache et conserver des DataFrames

Mise en cache : stocke les données en mémoire pour un accès plus rapide aux petits ensembles
Persistance : stocke selon différents niveaux pour les ensembles plus volumineux

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

# Mettre le DataFrame en cache
df.cache()


# Effectuer plusieurs opérations sur le DataFrame en cache
df.filter(df["column1"] > 50).show()
df.groupBy("column2").count().show()

Conserver des DataFrames à différents niveaux d'entreposage

# Conserver le DataFrame avec un niveau d'entreposage
from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)


# Effectuer des transformations
result = df.groupBy("column3").agg({"column4": "sum"})
result.show()

# Retirer de la persistance après usage
df.unpersist()

Optimiser PySpark

Petites sous‑sections : plus on utilise de données, plus l'opération ralentit ; privilégiez des outils comme map() plutôt que groupby() selon la sélectivité
Jointures « broadcast » : « broadcast » exploite tout le calcul, même pour de petits ensembles
Éviter de répéter les actions : répéter des actions sur les mêmes données coûte du temps et du calcul sans bénéfice

Passons à la pratique !

Introduction à PySpark