Agrégations PySpark

Introduction à PySpark

Benjamin Schmidt

Data Engineer

Aperçu des agrégations PySpark SQL

Les agrégations SQL courantes fonctionnent avec spark.sql()

# Requête d'agrégation SQL
spark.sql("""
  SELECT Department, SUM(Salary) AS Total_Salary, AVG(Salary) AS Average_Salary
  FROM employees
  GROUP BY Department
""").show()

Combiner DataFrame et opérations SQL

# Filtrer les salaires > 3000
filtered_df = df.filter(df.Salary > 3000)

# Enregistrer le DataFrame filtré comme vue
filtered_df.createOrReplaceTempView("filtered_employees")


# Agréger en SQL sur la vue filtrée
spark.sql("""
    SELECT Department, COUNT(*) AS Employee_Count
    FROM filtered_employees
    GROUP BY Department
""").show()

Gérer les types de données lors des agrégations

# Exemple de conversion de type
data = [("HR", "3000"), ("IT", "4000"), ("Finance", "3500")]
columns = ["Department", "Salary"]
df = spark.createDataFrame(data, schema=columns)


# Convertir la colonne Salary en entier
df = df.withColumn("Salary", df["Salary"].cast("int"))

# Effectuer l'agrégation
df.groupBy("Department").sum("Salary").show()

RDD pour les agrégations

# Exemple d'agrégation avec des RDD
rdd = df.rdd.map(lambda row: (row["Department"], row["Salary"]))


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


print(rdd_aggregated.collect())

Bonnes pratiques pour les agrégations PySpark

Filtrer tôt : réduire la taille avant d'agréger
Gérer les types : données propres et bien typées
Éviter les opérations sur l'ensemble complet : minimiser groupBy()
Choisir la bonne interface : privilégier les DataFrames pour leurs optimisations
Surveiller la performance : utiliser explain() pour le plan d'exécution et optimiser

Points clés

Agrégations PySpark SQL : fonctions comme SUM() et AVERAGE() pour résumer les données
DataFrames et SQL : combiner les deux pour une manipulation souple
Gestion des types : résoudre les incompatibilités de type lors des agrégations
RDD vs DataFrames : comprendre les compromis et choisir le bon outil

Passons à la pratique !

Introduction à PySpark