Batch-Verarbeitungsoptimierer

Was macht Batch-Verarbeitungsoptimierer?

Optimieren Sie Stapelverarbeitungs-Workloads mit Spark, Polars, DuckDB und pandas – mit Fokus auf Speichereffizienz, Parallelität und Kostensenkung.

Aktivierungsauslöser

Aktivieren bei: "batch processing", "Spark optimization", "Polars", "DuckDB", "pandas performance", "data frame", "shuffle optimization", "partition skew", "memory optimization", "out of memory"

Nicht für: Echtzeit-Streaming → streaming-pipeline-architect | Warehouse-SQL-Tuning → data-warehouse-optimizer | Pipeline-Orchestrierung → airflow-dag-orchestrator

Schnellstart

1. Wählen Sie das richtige Werkzeug – DuckDB für Single-Node-Analysen, Polars für DataFrames, Spark für verteilte Verarbeitung
2. Profilieren Sie zuerst – Identifizieren Sie Engpässe (Shuffle, Skew, Speicher) vor der Optimierung
3. Reduzieren Sie Daten frühzeitig – Filtern und wählen Sie Spalten so früh wie möglich in der Pipeline
4. Vermeiden Sie Shuffles – Broadcasten Sie kleine Tabellen, partitionieren Sie Daten vorab, nutzen Sie Map-Side-Joins
5. Dimensionieren Sie Ressourcen richtig – Passen Sie Executor-Speicher/Kerne an die tatsächliche Datengröße an

Kernfähigkeiten

Bereich	Technologien
Verteilt	Apache Spark 3.5+, Dask, Ray
Einzelknoten	DuckDB 1.1+, Polars 1.x, pandas 2.2+
Dateiformate	Parquet, Arrow IPC, Delta Lake, Iceberg
Optimierung	AQE (Spark), Lazy Evaluation (Polars), spaltenweise Scans
Cloud	Databricks, EMR, Dataproc, serverless Spark

Architekturmuster

Entscheidungsbaum zur Werkzeugauswahl

Datengröße?
  ├─ < 10 GB     → DuckDB (SQL) oder Polars (DataFrame)
  │                 Einzelner Rechner, keine Einrichtung, schnellste Iteration
  │
  ├─ 10-100 GB   → Polars (lazy) oder DuckDB (Out-of-Core)
  │                 Immer noch ein einzelner Rechner mit Spill-to-Disk
  │
  └─ > 100 GB    → Spark (verteilt)
                    Mehrknoten-Cluster, Shuffle-basierte Joins

Komplexität?
  ├─ SQL-zentriert  → DuckDB (schnellste SQL-Engine für Analysen)
  ├─ DataFrame      → Polars (10x schneller als pandas, Lazy Evaluation)
  └─ Komplexes ML   → Spark + MLlib oder Spark + Ray

Spark-Optimierungsmuster

from pyspark.sql import SparkSession
import pyspark.sql.functions as F

spark = SparkSession.builder \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .getOrCreate()

# GUT: kleine Dimensionstabelle broadcasten (< 100MB)
from pyspark.sql.functions import broadcast
result = large_df.join(broadcast(small_dim_df), "key")

# GUT: Prädikat-Pushdown – vor dem Join filtern
orders = spark.read.parquet("s3://data/orders/") \
    .filter(F.col("order_date") >= "2026-01-01") \
    .select("order_id", "customer_id", "amount")  # Spaltenauswahl

# SCHLECHT: collect() auf großen Datensätzen – verursacht OOM auf dem Treiber
# all_data = large_df.collect()  # NIEMALS so machen

# GUT: partitionierte Ausgabe schreiben
result.repartition(200) \
    .write.mode("overwrite") \
    .partitionBy("order_date") \
    .parquet("s3://output/results/")

Polars Lazy Evaluation

import polars as pl

# Lazy-Modus: erstellt Abfrageplan, optimiert, führt dann aus
result = (
    pl.scan_parquet("data/orders/*.parquet")  # Lazy Scan
    .filter(pl.col("order_date") >= "2026-01-01")
    .join(
        pl.scan_parquet("data/customers/*.parquet"),
        on="customer_id",
        how="inner"
    )
    .group_by("region")
    .agg([
        pl.col("amount").sum().alias("total_revenue"),
        pl.col("order_id").n_unique().alias("order_count"),
    ])
    .sort("total_revenue", descending=True)
    .collect()  # führt optimierten Plan aus
)

# Polars optimiert: Prädikat-Pushdown, Projektions-Pushdown,
# Join-Neuordnung – alles automatisch durch Lazy Evaluation

Anti-Patterns

1. pandas für >5GB – pandas lädt alles in den Speicher; verwenden Sie Polars (lazy) oder DuckDB für mittlere Daten, Spark für große
2. Collect zum Treiber – df.collect() oder df.toPandas() auf großen Spark-DataFrames verursacht OOM; aggregieren Sie zuerst
3. Partitions-Skew ignorieren – eine Partition mit 10x mehr Daten als andere bremst den gesamten Job aus; verwenden Sie AQE oder Salting
4. Alle Spalten lesen – wählen Sie immer nur benötigte Spalten aus; das Parquet-Spaltenformat überspringt ungenutzte Spalten vollständig
5. Zu kleine Ausgabedateien – zu viele kleine Ausgabedateien (< 128MB) verlangsamen nachgelagerte Lesevorgänge; führen Sie vor dem Schreiben ein Coalesce durch

Qualitäts-Checkliste

• Werkzeug passt zur Datengröße (DuckDB/Polars < 100GB, Spark > 100GB)
• Spalten frühzeitig reduziert (nur benötigte auswählen)
• Filter auf Scan-Ebene durchgeschleust (Prädikat-Pushdown)
• Kleine Tabellen in Joins gebroadcastet (< 100MB)
• Spark AQE aktiviert (Adaptive Query Execution)
• Kein collect() auf großen Datensätzen (vor dem Collect aggregieren)
• Ausgabedateien 128MB-1GB groß (Coalesce/Repartition vor dem Schreiben)
• Partitions-Skew überwacht und gemildert (Salting oder AQE)
• Job profiliert: Spark UI Stages, Polars .explain(), DuckDB EXPLAIN ANALYZE
• Speicher angemessen dimensioniert: Executor-Speicher >= 2x größte Partition