PySpark im großen Stil

Einführung in PySpark

Benjamin Schmidt

Data Engineer

Skalierung nutzen

PySpark arbeitet effektiv mit Gigabytes und Terabytes an Daten
Mit PySpark sind Geschwindigkeit und effiziente Verarbeitung das Ziel
PySpark-Ausführung verstehen bringt noch mehr Effizienz
Nutze Broadcast, um den gesamten Cluster zu verwalten

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

Ausführungspläne

# Verwende explain(), um den Ausführungsplan zu sehen
df.filter(df.Age > 40).select("Name").explain()

== Physischer 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 und persistieren

Caching: Speichert Daten im Speicher für schnelleren Zugriff bei kleineren Datensätzen
Persistieren: Speichert Daten auf verschiedenen Ebenen für größere Datensätze

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

# Cache das DataFrame
df.cache()


# Führe mehrere Operationen auf dem gecachten DataFrame aus
df.filter(df["column1"] > 50).show()
df.groupBy("column2").count().show()

DataFrames mit verschiedenen Speicherstufen persistieren

# Persistiere das DataFrame mit Speicherstufe
from pyspark import StorageLevel

df.persist(StorageLevel.MEMORY_AND_DISK)


# Transformationen durchführen
result = df.groupBy("column3").agg({"column4": "sum"})
result.show()

# Nach Gebrauch unpersistieren
df.unpersist()

PySpark optimieren

Kleine Unterabschnitte: Je mehr Daten, desto langsamer die Operation: Wähle Tools wie map() statt groupby() wegen der Selektivität
Broadcast-Joins: Broadcast nutzt alle Ressourcen, auch bei kleineren Datensätzen
Wiederholte Aktionen vermeiden: Wiederholte Aktionen auf denselben Daten kosten Zeit und Ressourcen ohne Nutzen

Lass uns üben!

Einführung in PySpark