स्केल पर PySpark

PySpark परिचय

Benjamin Schmidt

Data Engineer

स्केल का लाभ उठाएँ

PySpark गीगाबाइट और टेराबाइट डेटा पर प्रभावी रहता है
PySpark से लक्ष्य है तेज और कुशल प्रोसेसिंग
PySpark के execution को समझकर और कुशलता पाएँ
पूरे क्लस्टर को प्रबंधित करने के लिए broadcast का उपयोग करें

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

Execution plans

# execution plan देखने के लिए explain() का उपयोग
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

DataFrames को cache और persist करना

Caching: डेटा को मेमोरी में रखता है, छोटे डेटासेट पर तेज एक्सेस के लिए
Persisting: बड़े डेटासेट के लिए अलग-अलग स्टोरेज लेवल पर डेटा रखता है

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

# DataFrame को cache करें
df.cache()


# cached DataFrame पर कई ऑपरेशन चलाएँ
df.filter(df["column1"] > 50).show()
df.groupBy("column2").count().show()

विभिन्न स्टोरेज लेवल के साथ DataFrames persist करना

# स्टोरेज लेवल के साथ DataFrame को persist करें
from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)


# Transformations चलाएँ
result = df.groupBy("column3").agg({"column4": "sum"})
result.show()

# उपयोग के बाद unpersist करें
df.unpersist()

PySpark को ऑप्टिमाइज़ करना

छोटे सबसेक्शन: जितना अधिक डेटा उपयोग होगा, ऑपरेशन उतना धीमा होगा। चयनात्मकता के कारण groupby() की बजाय map() जैसे टूल चुनें
Broadcast Joins: broadcast छोटे डेटासेट पर भी सभी compute का उपयोग करता है
दोहराए गए actions से बचें: एक ही डेटा पर बार-बार action समय और compute खर्च करता है, बिना लाभ के

अभ्यास करते हैं!

PySpark परिचय