Efektywna agregacja i łączenie danych

Data Transformation with Spark SQL in Databricks

Disha Mukherjee

Lead Data Engineer

Trzy pytania, jeden przejrzysty zbiór danych

recraft: half: A data analyst reviewing colorful charts and business reports on multiple screens, isolated on transparent background

Które kategorie generują największe przychody?
Kim są najważniejsi klienci?
Jak wzbogacić wyniki o kontekst działu?

Jak działają groupBy() i agg()

groupBy() - dzieli wiersze według kolumny klucza
agg() - stosuje funkcje do każdej grupy równolegle
Obie operacje są leniwe — obliczenia uruchamia akcja
Jedno wywołanie agg() = jedno skanowanie danych

recraft: half: Colorful rows of data split into three separate labeled groups with a total value computed below each group, isolated on transparent background

Przychody według kategorii — 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())

)

Przychody według kategorii — wynik

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

Wzbogacanie danych o tabelę wymiarów

recraft: half: Two database tables being connected by a glowing link or bridge, isolated on transparent background

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

Standardowe złączenie lewostronne

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

Ukryty koszt — shuffle

recraft: half: Data packets flowing across a network of connected servers and nodes with arrows showing movement, isolated on transparent background

Pasujące klucze muszą trafić na tę samą maszynę
Spark przesyła wiersze przez sieć, aby je wyrównać
Shuffle = wąskie gardło przy dużej skali

Spark UI

recraft: full: A dark analytics monitoring dashboard with colorful horizontal bar charts showing job stages, progress bars, and data throughput metrics, isolated on transparent background

Jobs i Stages — sprawdź, co działało i jak długo
Shuffle Read/Write — ile danych zostało przesłanych
Przydatne przy diagnozowaniu wąskich gardeł potoku

Odczytywanie planu zapytania za pomocą .explain()

df_joined.explain(mode="formatted")

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

.explain() - analizuje plan wykonania
Węzły 15 i 16 - brak zarejestrowanych argumentów routingu

Złączenie broadcast — rozwiązanie

# 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

Przed i po — weryfikacja planu

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]

Tabela wymiarów ma instrukcje routingu
Duża tabela df_valid nie jest przesyłana
Jedna zmiana = minuty zamieniają się w sekundy przy dużej skali

Czas na praktykę!

Data Transformation with Spark SQL in Databricks