PySpark dalam skala besar

Pengantar PySpark

Benjamin Schmidt

Data Engineer

Memanfaatkan skala

PySpark bekerja efektif dengan data berukuran gigabyte dan terabyte
Menggunakan PySpark, kecepatan dan efisiensi pemrosesan adalah tujuannya
Memahami eksekusi PySpark memberikan efisiensi lebih
Gunakan broadcast untuk mengelola seluruh kluster

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

Rencana eksekusi

# Menggunakan explain() untuk melihat rencana eksekusi
df.filter(df.Age > 40).select("Name").explain()

== Rencana Fisik ==
*(1) Filter (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

Caching dan persisting DataFrames

Caching: Menyimpan data di memori, untuk akses lebih cepat pada dataset kecil
Persisting: Menyimpan data di berbagai level penyimpanan untuk dataset besar

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

# Cache DataFrame
df.cache()


# Lakukan beberapa operasi pada DataFrame yang di-cache
df.filter(df["column1"] > 50).show()
df.groupBy("column2").count().show()

Persisting DataFrames dengan level penyimpanan berbeda

# Persist DataFrame dengan level penyimpanan
from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)


# Lakukan transformasi
result = df.groupBy("column3").agg({"column4": "sum"})
result.show()

# Unpersist setelah digunakan
df.unpersist()

Mengoptimalkan PySpark

Subbagian Kecil: Semakin banyak data yang digunakan, semakin lambat operasinya: Pilih alat seperti map() daripada groupby() karena selektivitas metode
Broadcast Joins: Broadcast akan menggunakan semua komputasi, bahkan pada dataset kecil
Hindari Tindakan Berulang: Tindakan berulang pada data yang sama memakan waktu dan komputasi, tanpa manfaat

Mari berlatih!

Pengantar PySpark