Verteilte Big-Data-Analyse: Apache PySpark Architektur & Log-Analytics unter Linux

Erstellt: 2026-11-08  ·  Zuletzt aktualisiert: 2026-11-08  ·  Nejat Hakan

Management Summary: Wachsen Datensätze von Gigabytes in den Terabyte-Bereich, brechen Single-Node-Python-Prozesse durch RAM-Erschöpfung ab. Apache Spark überwindet die Hardwaregrenzen einzelner Server, indem es Berechnungen über verteilte Cluster von Worker-Knoten auf Basis gerichteter azyklischer Graphen (DAG) parallelisiert. PySpark verbindet die intuitive Syntax von Python mit Sparks nativem JVM-Rechenkern, dem Catalyst Query Optimizer und der Tungsten Off-Heap Memory-Engine. Dieser Leitfaden führt durch die verteilte Architektur, erklärt den Unterschied zwischen Narrow- und Wide-Shuffles, demonstriert das Memory-Tuning auf Debian und verarbeitet Multi-Gigabyte Server-Logs im Clusterbetrieb.

• Driver & Executor Topologie

  ---

  Cluster-Koordination verstehen: Wie der Spark Driver Aufgaben serialisiert und über JVM-Executor-Prozesse auf Linux-Knoten verteilt.

• Catalyst & Tungsten Engines

  ---

  JVM-Garbage-Collection umgehen: Off-Heap Binärspeicher mit Project Tungsten nutzen und Abfragepläne über Catalyst optimieren.

• Narrow vs. Wide Shuffles

  ---

  Netzwerk-Flaschenhälse vermeiden: Lokale In-Memory-Operationen (filter, map) von teuren Shuffle-Operationen (groupBy, join) trennen.

• Verteilte Log-Analytics

  ---

  Mustererkennung im Terabyte-Maßstab: Regex-Parsing von Nginx-Access-Logs und Export in partitionierte Parquet-Dateien.

---

1. Apache Spark Cluster-Topologie

Spark trennt die zentrale Aufgabenkoordination strikt von der physischen Berechnung auf den Workern:

%%{init: {'flowchart': {'wrappingWidth': 350}}}%%
flowchart TB
    APP["PySpark Client-Skript<br/>(spark-submit)"]

    subgraph Driver ["Spark Driver-Prozess"]
        direction TB
        SC["SparkSession"]
        CAT["Catalyst Optimizer"]
        SCHED["Task Scheduler"]
        SC --> CAT --> SCHED
    end

    subgraph Cluster ["Worker-Knoten"]
        direction TB
        subgraph W1 ["Worker-Knoten 1"]
            E1["Executor 1 (JVM)"]
        end
        subgraph W2 ["Worker-Knoten 2"]
            E2["Executor 2 (JVM)"]
        end
        W1 --> W2
    end

    STORAGE[("Verteilter Storage<br/>(HDFS / S3 / NVMe)")]

    APP --> Driver
    Driver ==> Cluster
    Cluster <==> STORAGE

Die Kernkomponenten:

• Spark Driver: Die Steuerzentrale mit der SparkSession. Sie übersetzt Python-Code in einen physischen Ausführungs-DAG und steuert die Executors.
• Worker-Knoten & Executors: Autonome JVM-Prozesse auf den Linux-Maschinen. Jeder Executor verfügt über einen Thread-Pool, der zugewiesene Datenpartitionen parallel verarbeitet.
• Tungsten Engine: Verwaltet Daten direkt in rohem C-ähnlichem Off-Heap-Speicher, wodurch der Java-Objekt-Overhead und störende Garbage-Collection-Pausen (GC) vollständig entfallen.

---

2. Lazy Evaluation & Die Kosten des Shufflings

PySpark-Code rechnet nicht sofort bei Funktionsaufruf: - Transformationen (Lazy): Befehle wie .select(), .filter() und .withColumn() ergänzen lediglich Anweisungen im Ausführungsplan. - Aktionen (Eager): Befehle wie .count(), .collect(), .show() oder .write.parquet() kompilieren den Plan und starten die physische Berechnung auf dem Cluster.

%%{init: {'flowchart': {'wrappingWidth': 350}}}%%
flowchart TB
    subgraph Narrow ["Narrow Transform (Schnell, Lokal)"]
        direction TB
        P1["Partition A"] -->|"filter / map"| O1["Ausgabe A (Gleicher Worker)"]
    end

    subgraph Wide ["Wide Transform (Netzwerk-Shuffle)"]
        direction TB
        W_P["Partitionen 1 & 2"]
        SHUFFLE["Netzwerk-Shuffle<br/>(Disk I/O + Serialisierung)"]
        W_O["Aggregierte Ausgaben"]
        W_P --> SHUFFLE --> W_O
    end

    Narrow -->|"Erfordert Netzwerk-Transfer"| Wide

Vorsicht vor Wide Transformations bei großen Datenmengen!

Operationen, die Daten über Partitionsgrenzen hinweg neu anordnen (groupByKey, Joins), erzwingen einen Netzwerk-Shuffle. Daten werden auf lokale NVMe-Platten geschrieben, über das Netzwerk übertragen und neu sortiert. Filtern Sie Daten daher immer so früh wie möglich vor.

---

3. Ressourcen-Tuning unter Debian Linux

Übergeben Sie Produktiv-Jobs über spark-submit mit dedizierter Ressourcenallokation:

# Datei: /opt/ds-ops/submit_spark_job.sh
spark-submit \
    --master spark://cluster-master:7077 \
    --deploy-mode cluster \
    --driver-memory 4G \
    --executor-memory 16G \
    --executor-cores 4 \
    --num-executors 10 \
    --conf spark.sql.shuffle.partitions=200 \
    --conf spark.memory.fraction=0.8 \
    --conf spark.memory.storageFraction=0.3 \
    /opt/ds-ops/pyspark_log_analytics.py

Faustregeln zur Dimensionierung:

1. Core-Allokation: Weisen Sie jedem Executor 4 bis 5 Cores zu. Mehr als 5 Cores führen zu I/O-Engpässen innerhalb einer einzelnen JVM.
2. Shuffle-Partitionen (spark.sql.shuffle.partitions): Standard ist 200. Zielgröße: 100–200 MB pro Partition. Bei 100 GB Shuffle-Volumen: 100.000 MB / 150 MB ≈ 650 Partitionen.

---

4. Praxis-Lab: Verteilte Nginx-Log-Analyse

Führen Sie dieses Skript aus, um Millionen von Nginx-Access-Logzeilen über alle verfügbaren Kerne hinweg strukturiert zu zerlegen:

#!/usr/bin/env python3
# Datei: /opt/ds-ops/pyspark_log_analytics.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col,
    count,
    desc,
    hour,
    regexp_extract,
    to_timestamp,
    when
)


def init_spark_session() -> SparkSession:
    """Initialisiert eine optimierte lokale Spark-Session."""
    return (
        SparkSession.builder
        .appName("EnterpriseLogAnalytics")
        .master("local[*]")
        .config("spark.sql.shuffle.partitions", "32")
        .config("spark.driver.memory", "8g")
        .config("spark.sql.adaptive.enabled", "true")
        .getOrCreate()
    )


def main():
    spark = init_spark_session()
    spark.sparkContext.setLogLevel("WARN")
    print("Spark Session erfolgreich initialisiert.")

    # 1. Regex-Pattern für Nginx Combined Log Format
    log_pattern = r'^(\S+) \S+ \S+ \[([\w:/]+\s[+\-]\d{4})\] "(\S+) (\S+)\s*\S*" (\d{3}) (\d+)'

    log_path = "/var/log/nginx/access.log"
    print(f"Lese Rohdaten-Stream: {log_path}")

    # 2. Strukturierte Extraktions-Pipeline
    raw_df = spark.read.text(log_path)

    parsed_df = (
        raw_df
        .select(
            regexp_extract("value", log_pattern, 1).alias("client_ip"),
            regexp_extract("value", log_pattern, 2).alias("raw_time"),
            regexp_extract("value", log_pattern, 3).alias("http_method"),
            regexp_extract("value", log_pattern, 4).alias("endpoint"),
            regexp_extract("value", log_pattern, 5).cast("integer").alias("status_code"),
            regexp_extract("value", log_pattern, 6).cast("long").alias("response_bytes")
        )
        .filter(col("client_ip") != "")
        .withColumn("is_error", when(col("status_code") >= 400, 1).otherwise(0))
    )

    # 3. Performante Aggregationen via Catalyst
    print("\n=== Top 5 Fehler-Endpunkte (HTTP 4xx / 5xx) ===")
    error_summary = (
        parsed_df
        .filter(col("is_error") == 1)
        .groupBy("endpoint", "status_code")
        .agg(count("*").alias("error_count"))
        .orderBy(desc("error_count"))
        .limit(5)
    )
    error_summary.show(truncate=False)

    print("\n=== Top 5 bandbreitenintensive Client-IPs ===")
    (
        parsed_df
        .groupBy("client_ip")
        .sum("response_bytes")
        .orderBy(desc("sum(response_bytes)"))
        .limit(5)
        .show()
    )

    # 4. Export in partitioniertes Parquet-Format
    output_target = "/var/data/analytics/nginx_parsed_parquet"
    print(f"Speichere partitioniertes Parquet unter: {output_target}")
    (
        parsed_df
        .write
        .mode("overwrite")
        .partitionBy("status_code")
        .parquet(output_target)
    )

    print("[OK] Verteilte Log-Analyse erfolgreich abgeschlossen.")
    spark.stop()


if __name__ == "__main__":
    main()