Performans iyileştirmeleri

PySpark ile Veri Temizleme

Mike Metzger

Data Engineering Consultant

Spark çalışma planını açıklama

voter_df = df.select(df['VOTER NAME']).distinct()
voter_df.explain()

== Physical Plan ==
*(2) HashAggregate(keys=[VOTER NAME#15], functions=[])
+- Exchange hashpartitioning(VOTER NAME#15, 200)
   +- *(1) HashAggregate(keys=[VOTER NAME#15], functions=[])
      +- *(1) FileScan csv [VOTER NAME#15] Batched: false, Format: CSV, Location: 
      InMemoryFileIndex[file:/DallasCouncilVotes.csv.gz], 
      PartitionFilters: [], PushedFilters: [], 
      ReadSchema: struct<VOTER NAME:string>

Shuffle nedir?

Shuffle (karıştırma), bir görevi tamamlamak için verinin farklı işçilere taşınmasıdır

Karmaşıklığı kullanıcıdan gizler
Tamamlanması yavaş olabilir
Genel verimi düşürür
Genellikle gereklidir; yine de en aza indirin

Shuffle nasıl sınırlandırılır?

.repartition(num_partitions) kullanımını sınırlayın
- Bunun yerine .coalesce(num_partitions) kullanın
.join() çağırırken dikkatli olun
.broadcast() kullanın
Her zaman sınırlamak gerekmeyebilir

Broadcast

Broadcast (yayınlama):

Her işçiye bir nesnenin kopyasını verir
Düğümler arası gereksiz iletişimi önler
.join() işlemlerini ciddi biçimde hızlandırabilir

.broadcast(<DataFrame>) yöntemini kullanın

from pyspark.sql.functions import broadcast
combined_df = df_1.join(broadcast(df_2))

Hadi pratik yapalım!

PySpark ile Veri Temizleme