Verileri verimli şekilde birleştirme ve özetleme

Databricks'te Spark SQL ile Veri Dönüştürme

Disha Mukherjee

Lead Data Engineer

Üç soru, tek temiz veri kümesi

recraft: half: Bir veri analisti, şeffaf arka planda birden fazla ekrandaki renkli grafik ve iş raporlarını inceliyor

En çok geliri hangi kategoriler getiriyor?
En iyi müşteriler kimler?
Sonuçları departman bağlamıyla nasıl zenginleştiririz?

groupBy() ve agg() nasıl çalışır

groupBy() - satırları anahtar sütuna göre böler
agg() - fonksiyonları gruplara paralel uygular
İkisi de tembel - işlem bir eylemle tetiklenir
Tek agg() çağrısı = tek veri taraması

recraft: half: Renkli veri satırları üç ayrı etiketli gruba bölünmüş ve her grubun altında toplam değer hesaplanmış, şeffaf arka planda izole

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

)

Kategori geliri - çıktı

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

Boyut tablosuyla zenginleştirme

recraft: half: Parlayan bir bağlantı veya köprüyle birbirine bağlanan iki veritabanı tablosu, şeffaf arka planda izole

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

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

Gizli maliyet - shuffle

recraft: half: Bağlı sunucular ve düğümler ağında hareketi gösteren oklarla akan veri paketleri, şeffaf arka planda izole

Eşleşen anahtarlar aynı makineye düşmeli
Spark, hizalamak için satırları ağda taşır
Shuffle = ölçekta dar boğaz

Spark UI

recraft: full: İş aşamalarını, ilerleme çubuklarını ve veri aktarım metriklerini gösteren renkli yatay çubuk grafiklere sahip koyu bir analiz izleme paneli, şeffaf arka planda izole

Jobs ve Stages - ne koştu ve ne kadar sürdü izlenir
Shuffle Read/Write - ne kadar veri taşındı görürsün
Boru hattı dar boğazlarını teşhis için faydalı

.explain() ile sorgu planını okuma

df_joined.explain(mode="formatted")

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

.explain() - çalışma planını incele
Düğümler 15 ve 16 - yönlendirme Argümanları yok kaydedildi

Broadcast join - çözüm

# Küçük tabloyu F.broadcast() ile sarmala
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

Öncesi ve sonrası - planı doğrulama

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]

Boyut tablosunda yönlendirme talimatları var
Büyük tablo df_valid asla hareket etmez
Tek değişiklik = ölçekte dakikadan saniyeye

Hadi pratik yapalım!

Databricks'te Spark SQL ile Veri Dönüştürme