Mehr zu Spark DataFrames

Einführung in PySpark

Benjamin Schmidt

Data Engineer

Erstellen von DataFrames aus verschiedenen Datenquellen

CSV-Dateien: Für strukturierte, separierte Daten
JSON Files: Halbstrukturiertes, hierarchisches Datenformat
Parquet-Dateien: Optimiert für die Speicherung und Abfrage, wird oft im Bereich Data Engineering genutzt.

Beispiel:
```
spark.read.csv("path/to/file.csv")
```
Beispiel:
```
spark.read.json("path/to/file.json")
```

Beispiel:

spark.read.parquet("path/to/file.parquet")

¹ https://spark.apache.org/docs/latest/api/python/reference/pyspark.pandas/api/pyspark.pandas.read_csv

Schema-Ableitung und manuelle Schema-Definition

Spark kann Schemata aus Daten ableiten mit inferSchema=True
Schema manuell festlegen, um mehr Kontrolle zu haben – super für fixe Datenstrukturen

Schema at Scale

Datentypen in PySpark-DataFrames

IntegerType: Ganze Zahlen
- Z. B.: 1, 3478, -1890456
LongType: Größere ganze Zahlen
- Zum Beispiel 8-Byte-Zahlen mit Vorzeichen: 922334775806
FloatType und DoubleType: Gleitkommazahlen für Dezimalwerte
- Z. B.: 3.14159
StringType: Wird für Text- oder Zeichenfolgen-Daten benutzt
- Z. B.: "This is an example of a string."
...

Datentypen-Syntax für PySpark-DataFrames

# Import the necessary types as classes
from pyspark.sql.types import (StructType,
                            StructField, IntegerType,
                            StringType, ArrayType)

# Construct the schema
schema = StructType([
    StructField("id", IntegerType(), True),
    StructField("name", StringType(), True),
    StructField("scores", ArrayType(IntegerType()), True)
])

# Set the schema
df = spark.createDataFrame(data, schema=schema)

DataFrame-Operationen – Auswahl und Filterung

Benutze .select(), um bestimmte Spalten auszuwählen.
Benutze .filter() oder .where(), um Zeilen nach Bedingungen zu filtern.
Benutze .sort(), um nach einer Reihe von Spalten zu sortieren.

# Select and show only the name and age columns
df.select("name", "age").show()

# Filter on age > 30
df.filter(df["age"] > 30).show()

# Use Where to filter match a specific value
df.where(df["age"] == 30).show()

# Use Sort to sort on age
df.sort("age", ascending=False).show()

Sortieren und Löschen fehlender Werte

Daten mittels .sort() oder .orderBy() sortieren
Benutze na.drop(), um Zeilen mit Nullwerten zu löschen.

# Sort using the age column
df.sort("age", ascending=False).show()

# Drop missing values
df.na.drop().show()

Spickzettel

spark.read_json(): Daten aus JSON laden
spark.read.schema(): Schemas explizit definieren
.na.drop(): Zeilen mit fehlenden Werten löschen
.select(), .filter(), .sort(), .orderBy(): Grundlegende Funktionen zur Datenbearbeitung

Lass uns üben!

Einführung in PySpark