Agrégations PySpark

Introduction à PySpark

Benjamin Schmidt

Data Engineer

Vue d'ensemble des agrégations SQL PySpark

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()

Combinaison des opérations DataFrame et SQL

# Filtrer les salaires supérieurs à 3000
filtered_df = df.filter(df.Salary > 3000)

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


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

Gestion des types de données dans les 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()

RDDs pour les agrégations

# Exemple d'agrégation avec RDDs
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 des données avant les agrégations
Gérer les types de données : Assurer la propreté et la bonne typage des données
Éviter les opérations sur l'ensemble du dataset : Minimiser les opérations comme groupBy()
Choisir la bonne interface : Préférer les DataFrames pour la plupart des tâches grâce à leurs optimisations
Surveiller la performance : Utiliser explain() pour inspecter le plan d'exécution et optimiser

Points clés

Agrégations SQL PySpark : Fonctions comme SUM() et AVERAGE() pour résumer les données
DataFrames et SQL : Combiner les deux approches pour une manipulation flexible des données
Gestion des types de données : Résoudre les problèmes de compatibilité de types lors des agrégations
RDDs vs DataFrames : Comprendre les compromis et choisir le bon outil

Passons à la pratique !

Introduction à PySpark