PySpark ในระดับขนาดใหญ่

PySpark เบื้องต้น

Benjamin Schmidt

Data Engineer

การใช้ประโยชน์จาก scale

PySpark ทำงานได้อย่างมีประสิทธิภาพกับข้อมูลระดับกิกะไบต์และเทราไบต์
เป้าหมายของ PySpark คือความเร็วและการประมวลผลที่มีประสิทธิภาพ
เข้าใจการทำงานของ PySpark เพื่อเพิ่มประสิทธิภาพยิ่งขึ้น
ใช้ broadcast เพื่อจัดการทั้งคลัสเตอร์

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

Execution plan

# Using explain() to view the execution plan
df.filter(df.Age > 40).select("Name").explain()

== Physical Plan ==
*(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

การ Cache และ Persist DataFrame

Caching: เก็บข้อมูลไว้ใน memory เพื่อการเข้าถึงที่รวดเร็วขึ้น เหมาะกับชุดข้อมูลขนาดเล็ก
Persisting: เก็บข้อมูลในระดับ storage ที่ต่างกัน เหมาะกับชุดข้อมูลขนาดใหญ่

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

# Cache the DataFrame
df.cache()


# Perform multiple operations on the cached DataFrame
df.filter(df["column1"] > 50).show()
df.groupBy("column2").count().show()

การ Persist DataFrame ด้วย storage level ที่ต่างกัน

# Persist the DataFrame with storage level
from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)


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

# Unpersist after use
df.unpersist()

การ optimize PySpark

ใช้ข้อมูลส่วนย่อย: ยิ่งใช้ข้อมูลมากขึ้น การทำงานยิ่งช้าลง — เลือกใช้ map() แทน groupby() เพื่อความแม่นยำในการเลือกข้อมูล
Broadcast Joins: broadcast ใช้ทรัพยากรการคำนวณทั้งหมด แม้กับชุดข้อมูลขนาดเล็ก
หลีกเลี่ยงการรันซ้ำ: การรันซ้ำกับข้อมูลชุดเดิมเปลืองเวลาและทรัพยากรโดยไม่จำเป็น

มาฝึกกันเถอะ!

PySpark เบื้องต้น