Agréger et joindre des données efficacement

Transformation des données avec Spark SQL dans Databricks

Disha Mukherjee

Lead Data Engineer

Trois questions, un jeu de données propre

recraft: half: Un·e data analyst examinant des graphiques colorés et des rapports métier sur plusieurs écrans, sur fond transparent

Quelles catégories génèrent le plus de revenus ?
Qui sont les meilleurs clients ?
Comment enrichir les résultats avec le service ?

Fonctionnement de groupBy() et agg()

groupBy() : partitionne les lignes par colonne clé
agg() : applique des fonctions à chaque groupe en parallèle
Les deux sont lazies : une action déclenche le calcul
Un appel à agg() = un seul scan des données

recraft: half: Lignes de données colorées divisées en trois groupes étiquetés avec une valeur totale calculée sous chaque groupe, sur fond transparent

Revenus par catégorie – code

category_revenue = (
    df_valid
    .groupBy("Category")

    .agg(
        F.round(F.sum("Transaction_Amount"), 2).alias("total_revenue"),
        F.count("Transaction_Amount").alias("transaction_count"),
        F.round(F.avg("Transaction_Amount"), 2).alias("avg_transaction"),
    )

    .orderBy(F.col("total_revenue").desc())

)

Revenus par catégorie – résultat

category_revenue.show(truncate=False)

+-----------+-----------------+-----------------+---------------+
|Category   |total_revenue    |transaction_count|avg_transaction|
+-----------+-----------------+-----------------+---------------+
|Clothing   |338266605.24     |6742             |50173.04       |
|Dining     |337631718.64     |6719             |50250.29       |
|Electronics|331102843.72     |6614             |50060.91       |
|Savings    |329346609.07     |6596             |49931.26       |
|Groceries  |329003451.57     |6537             |50329.43       |
|Unknown    |880591.27        |15               |58706.08       |
+-----------+-----------------+-----------------+---------------+

Enrichir avec une table de dimensions

recraft: half: Deux tables de base de données reliées par un lien lumineux, sur fond transparent

+-----------+----------+
|Category   |Department|
+-----------+----------+
|Clothing   |Retail    |
|Dining     |Food      |
|Electronics|Tech      |
|Groceries  |Food      |
|Savings    |Finance   |
+-----------+----------+

Jointure gauche standard

df_joined = df_valid.join(df_dim, on="Category", how="left")


df_joined.select(
    "Customer_ID", "Category", "Department", "Transaction_Amount"
).show(5, truncate=False)

+-----------+-----------+----------+------------------+
|Customer_ID|Category   |Department|Transaction_Amount|
+-----------+-----------+----------+------------------+
|CUST003    |Electronics|Tech      |5752.36           |
|CUST009    |Clothing   |Retail    |28959.12          |
|CUST010    |Savings    |Finance   |72098.18          |
|CUST011    |Savings    |Finance   |49771.05          |
|CUST020    |Groceries  |Food      |69825.14          |
+-----------+-----------+----------+------------------+

Le coût caché : le shuffle

recraft: half: Paquets de données circulant sur un réseau de serveurs et nœuds connectés avec des flèches indiquant le flux, sur fond transparent

Les clés correspondantes doivent arriver sur la même machine
Spark déplace des lignes sur le réseau pour les aligner
Shuffle = goulot d'étranglement à grande échelle

Spark UI

recraft: full: Un tableau de bord d'analyse sombre avec des diagrammes à barres horizontales colorés montrant les étapes de job, barres de progression et débits, sur fond transparent

Jobs et Stages : suivre l'exécution et la durée
Shuffle Read/Write : quantifier les données déplacées
Utile pour diagnostiquer les goulots d'étranglement de pipeline

Lire le plan de requête avec .explain()

df_joined.explain(mode="formatted")

...
+- PhotonBroadcastHashJoin LeftOuter (18)
   :- ...
   +- PhotonShuffleExchangeSource (17)
      +- PhotonShuffleMapStage (16)
         +- PhotonShuffleExchangeSink (15)
            +- LocalTableScan (13)
...

.explain() : inspecter le plan d'exécution
Nœuds 15 et 16 : aucun Argument de routage journalisé

Broadcast join – la solution

# Envelopper la petite table avec F.broadcast()
df_broadcast = df_valid.join(
    F.broadcast(df_dim),
    on="Category",
    how="left"
)


print(f"Broadcast joined rows: {df_broadcast.count():,}")

Broadcast joined rows: 33,223

Avant/après – vérifier le plan

df_broadcast.explain(mode="formatted")

+- PhotonBroadcastHashJoin LeftOuter (18)
   :- ...
   +- PhotonShuffleExchangeSource (17)
      +- PhotonShuffleMapStage (16)
         +- PhotonShuffleExchangeSink (15)
            +- LocalTableScan (13)
...            
(15) PhotonShuffleExchangeSink
Arguments: SinglePartition
(16) PhotonShuffleMapStage
Arguments: EXECUTOR_BROADCAST, [id=#11839]

La table de dimensions a des instructions de routage
La grande table df_valid ne bouge jamais
Un changement = des minutes aux secondes à l'échelle

Passons à la pratique !

Transformation des données avec Spark SQL dans Databricks