PySpark ile Ölçekleme

PySpark'e Giriş

Benjamin Schmidt

Data Engineer

Ölçekten Yararlanma

PySpark, gigabayt ve terabayt boyutundaki verilerle etkili çalışır
PySpark kullanarak hız ve verimli işlem hedeflenir
PySpark yürütmesini anlamak daha fazla verimlilik sağlar
Tüm kümeyi yönetmek için yayını kullanın

joined_df = large_df.join(broadcast(small_df), 
                          on="key_column", how="inner")
joined_df.show()

Yürütme Planları

# Yürütme planını görmek için explain() kullanma
df.filter(df.Age > 40).select("Name").explain()

== Fiziksel Plan ==
*(1) Filtre (isnotnull(Age) AND (Age > 30))
+- Scan ExistingRDD[Name:String, Age:Int]

¹ https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.explain.html

DataFrame'leri Önbelleğe Alma ve Kalıcı Kılma

Önbelleğe Alma: Küçük veri kümeleri için daha hızlı erişim sağlamak üzere veriyi bellekte saklar
Kalıcı Kılma: Büyük veri kümeleri için farklı depolama seviyelerinde veriyi saklar

df = spark.read.csv("large_dataset.csv", header=True, inferSchema=True)

# DataFrame'i önbelleğe al
df.cache()


# Önbelleğe alınmış DataFrame üzerinde birden fazla işlem yap
df.filter(df["column1"] > 50).show()
df.groupBy("column2").count().show()

Farklı Depolama Seviyeleri ile DataFrame'leri Kalıcı Kılma

# Depolama seviyesi ile DataFrame'i kalıcı kılma
from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)


# Dönüşümleri gerçekleştir
result = df.groupBy("column3").agg({"column4": "sum"})
result.show()

# Kullanımdan sonra kalıcılığı kaldır
df.unpersist()

PySpark'ı Optimize Etme

Küçük Alt Bölümler: Daha fazla veri kullanıldıkça işlem yavaşlar: map() gibi seçici yöntemleri groupby() yerine tercih edin
Yayın Birleştirmeleri: Yayın, küçük veri kümelerinde bile tüm hesaplamayı kullanır
Tekrarlanan İşlemlerden Kaçının: Aynı veri üzerinde tekrarlanan işlemler zaman ve hesaplama maliyeti yaratır, fayda sağlamaz

Hadi pratik yapalım!

PySpark'e Giriş