Agregarea și îmbinarea eficientă a datelor

Transformarea datelor cu Spark SQL în Databricks

Disha Mukherjee

Lead Data Engineer

Trei întrebări, un set de date curat

recraft: half: Un analist de date analizând grafice colorate și rapoarte de business pe mai multe ecrane, izolat pe fundal transparent

Ce categorii generează cel mai mare venit?
Cine sunt clienții de top?
Cum îmbogățim rezultatele cu context departamental?

Cum funcționează `groupBy()` și `agg()`

groupBy() - împarte rândurile după coloana-cheie
agg() - aplică funcții fiecărui grup în paralel
Ambele sunt lazy - acțiunea declanșează calculul
Un singur apel agg() = o singură scanare a datelor

recraft: half: Rânduri colorate de date împărțite în trei grupuri distincte etichetate, cu o valoare totală calculată sub fiecare grup, izolat pe fundal transparent

Venituri pe categorie – cod

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

)

Venituri pe categorie – rezultat

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       |
+-----------+-----------------+-----------------+---------------+

Îmbogățirea cu un tabel de dimensiuni

recraft: half: Două tabele de baze de date conectate printr-o legătură sau punte luminoasă, izolat pe fundal transparent

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

Join stânga 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          |
+-----------+-----------+----------+------------------+

Costul ascuns – shuffle

recraft: half: Pachete de date care circulă printr-o rețea de servere și noduri conectate, cu săgeți care indică direcția, izolat pe fundal transparent

Cheile potrivite trebuie să ajungă pe același nod
Spark transferă rânduri prin rețea pentru a le alinia
Shuffle = blocaj la scară mare

Spark UI

recraft: full: Un panou de monitorizare analytics întunecat cu grafice orizontale colorate ce afișează etapele joburilor, bare de progres și metrici de transfer de date, izolat pe fundal transparent

Jobs și Stages – urmărește ce a rulat și cât timp
Shuffle Read/Write – vezi câte date au fost mutate
Util pentru diagnosticarea blocajelor din pipeline

Citirea planului de interogare cu `.explain()`

df_joined.explain(mode="formatted")

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

.explain() - inspectează planul de execuție
Nodurile 15 & 16 - nu există argumente de rutare înregistrate

Broadcast join – soluția

# Wrap the small table in 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

Înainte și după – verificarea planului

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]

Tabelul de dimensiuni are instrucțiuni de rutare
Tabelul mare df_valid nu se mișcă niciodată
O singură modificare = minute reduse la secunde la scară mare

Hai să exersăm!

Transformarea datelor cu Spark SQL în Databricks