Aggregera och joina data effektivt

Datatransformation med Spark SQL i Databricks

Disha Mukherjee

Lead Data Engineer

Tre frågor, ett rent dataset

recraft: half: En dataanalytiker granskar färgglada diagram och affärsrapporter på flera skärmar, isolerad på transparent bakgrund

Vilka kategorier genererar mest intäkter?
Vilka är de toppkunder?
Hur berikar vi resultaten med avdelningskontext?

Så fungerar groupBy() och agg()

groupBy() - delar upp rader efter nyckelkolumn
agg() - tillämpar funktioner på varje grupp parallellt
Båda är lata – en action utlöser beräkningen
Ett agg()-anrop = en datasökning

recraft: half: Färgglada datarader uppdelade i tre separata märkta grupper med ett totalvärde beräknat under varje grupp, isolerade på transparent bakgrund

Kategoriintäkter – kod

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

)

Kategoriintäkter – utdata

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

Berika med en dimensionstabell

recraft: half: Två databastabeller kopplade samman via en lysande länk eller brygga, isolerade på transparent bakgrund

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

Vanlig vänsterjoin

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

Den dolda kostnaden – shuffle

recraft: half: Datapaket som flödar genom ett nätverk av anslutna servrar och noder med pilar som visar rörelseriktning, isolerade på transparent bakgrund

Matchande nycklar måste hamna på samma maskin
Spark skickar rader över nätverket för att rada upp dem
Shuffle = flaskhals i stor skala

Spark UI

recraft: full: En mörk analysdashboard med färgglada horisontella stapeldiagram som visar jobbfaser, förloppsindikatorer och dataflödesmätvärden, isolerad på transparent bakgrund

Jobs och Stages – spåra vad som kördes och hur länge
Shuffle Read/Write – se hur mycket data som flyttades
Användbart för att diagnostisera pipelineflaskhalsar

Läsa frågeplan med .explain()

df_joined.explain(mode="formatted")

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

.explain() – granska exekveringsplanen
Nod 15 & 16 – inga routningargument loggade

Broadcast-join – lösningen

# 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

Före och efter – verifiera planen

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]

Dimensionstabellen har routningsinstruktioner
Den stora tabellen df_valid flyttas aldrig
En ändring = minuter till sekunder i stor skala

Nu kör vi en övning!

Datatransformation med Spark SQL i Databricks