PySpark la scară

Introducere în PySpark

Benjamin Schmidt

Data Engineer

Valorificarea scalabilității

PySpark lucrează eficient cu gigaocteți și teraocteți de date
Folosind PySpark, scopul este viteza și procesarea eficientă
Înțelegerea execuției PySpark aduce eficiențe suplimentare
Utilizați broadcast pentru a gestiona întregul cluster

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

Planuri de execuție

# 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

Caching și persisting pentru DataFrame-uri

Caching: Stochează datele în memorie pentru acces mai rapid, pe seturi mici de date
Persisting: Stochează datele la diferite niveluri de stocare, pentru seturi mari de date

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()

Persisting DataFrame-uri cu niveluri diferite de stocare

# 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()

Optimizarea PySpark

Subsecțiuni mici: Cu cât mai multe date, cu atât operația e mai lentă: Folosiți metode precum map() în loc de groupby() pentru selectivitate
Join-uri Broadcast: Broadcast utilizează toată puterea de calcul, chiar și pe seturi mici de date
Evitați acțiunile repetate: Acțiunile repetate pe aceleași date consumă timp și resurse fără niciun beneficiu

Să exersăm!

Introducere în PySpark