PySpark op schaal

Introductie tot PySpark

Benjamin Schmidt

Data Engineer

Schaal benutten

PySpark werkt effectief met gigabytes en terabytes aan data
Met PySpark is snelheid en efficiëntie het doel
Begrip van PySpark-uitvoering levert nog meer efficiëntie op
Gebruik broadcast om het hele cluster te beheren

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

Uitvoeringsplannen

# Gebruik explain() om het uitvoeringsplan te bekijken
df.filter(df.Age > 40).select("Name").explain()

== Fysiek 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 cachen en persisten

Cachen: Slaat data op in het geheugen voor snellere toegang bij kleinere datasets
Persisten: Slaat data op in verschillende opslagniveaus voor grotere datasets

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

# Cache de DataFrame
df.cache()


# Voer meerdere operaties uit op de gecachte DataFrame
df.filter(df["column1"] > 50).show()
df.groupBy("column2").count().show()

DataFrames persisteren met verschillende opslagniveaus

# Persisteer de DataFrame met opslagniveau
from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)


# Voer transformaties uit
result = df.groupBy("column3").agg({"column4": "sum"})
result.show()

# Unpersist na gebruik
df.unpersist()

PySpark optimaliseren

Kleine secties: Hoe meer data, hoe trager de operatie: Kies tools zoals map() boven groupby() vanwege de selectiviteit
Broadcast Joins: Broadcast gebruikt alle rekenkracht, zelfs bij kleinere datasets
Vermijd herhaalde acties: Herhaalde acties op dezelfde data kosten tijd en rekenkracht zonder voordeel

Aan de slag!

Introductie tot PySpark