Data efficiënt aggregeren en joinen

Gegevens transformeren met Spark SQL in Databricks

Disha Mukherjee

Lead Data Engineer

Drie vragen, één schone gegevensset

recraft: half: Een data-analist die kleurrijke grafieken en bedrijfsrapporten op meerdere schermen bekijkt, geïsoleerd op transparante achtergrond

Welke categorieën leveren de meeste omzet op?
Wie zijn de topklanten?
Hoe verrijk je resultaten met de context van afdelingen?

Hoe groupBy() en agg() werken

groupBy() - verdeelt rijen op sleutelkolom
agg() - past functies per groep parallel toe
Beide zijn lui - actie start de berekening
Eén agg()-aanroep = één datascan

recraft: half: Kleurrijke rijen data verdeeld in drie afzonderlijke gelabelde groepen met onder elke groep een totaalwaarde, geïsoleerd op transparante achtergrond

Omzet per categorie - 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())

)

Omzet per categorie - output

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

Verrijken met een dimensietabel

recraft: half: Twee databasetabellen die worden verbonden door een gloeiende link of brug, geïsoleerd op transparante achtergrond

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

Standaard left join

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

De verborgen kosten - shuffle

recraft: half: Datapakketten die stromen over een netwerk van verbonden servers en nodes met pijlen die de beweging tonen, geïsoleerd op transparante achtergrond

Overeenkomende sleutels moeten op dezelfde machine landen
Spark verzendt rijen over het netwerk om ze uit te lijnen
Shuffle = bottleneck op schaal

Spark UI

recraft: full: Een donker analytics-monitoringdashboard met kleurrijke horizontale staafgrafieken met jobfases, voortgangsbalken en doorvoermetingen, geïsoleerd op transparante achtergrond

Jobs en Stages - zie wat draaide en hoe lang
Shuffle Read/Write - hoeveel data bewoog
Handig om pijplijnbottlenecks te vinden

Het queryplan lezen met .explain()

df_joined.explain(mode="formatted")

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

.explain() - bekijk het uitvoeringsplan
Nodes 15 & 16 - geen routing Arguments gelogd

Broadcast join - de oplossing

# Wikkel de kleine tabel 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

Voor en na - plan verifiëren

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]

Dim-tabel heeft routeringsinstructies
Grote tabel df_valid verplaatst nooit
Eén aanpassing = van minuten naar seconden op schaal

Laten we oefenen!

Gegevens transformeren met Spark SQL in Databricks