Production Machine Learning: Leckagefreie Scikit-learn Pipelines & ColumnTransformer

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

Management Summary: Über 90 % aller Machine-Learning-Modelle, die im Jupyter Notebook scheinbar 99 % Genauigkeit melden, versagen im Produktiveinsatz vollständig. Die häufigste Ursache ist Train-Test Data Leakage: Skalierungsfaktoren, Imputationsmediane oder One-Hot-Encoder werden vor dem Train/Test-Split auf dem Gesamtdatensatz berechnet. Dieser Leitfaden beschreibt den Aufbau robuster, datenleckfreier ML-Pipelines unter Debian Linux mittels Scikit-learns Pipeline und ColumnTransformer – inklusive atomarer Vorverarbeitung, robuster Behandlung ungesehener Kategorien und deterministischer joblib-Serialisierung.

• Leckagefreie Architektur

  ---

  Validierungs-Kontamination verhindern: Imputation, Skalierung und Modell-Fit atomar in einer Pipeline kapseln, die strikt nur auf Trainings-Folds gefittet wird.

• ColumnTransformer-Verzweigung

  ---

  Heterogene Merkmale parallel verarbeiten: RobustScaler für ausreißerbehaftete numerische Features und OneHotEncoder(handle_unknown='ignore') für Kategorien.

• Resilienz gegen Daten-Drift

  ---

  API-Abstürze im Live-Betrieb ausschließen: Sicherstellen, dass neue Kategorien oder fehlende Messwerte in REST-Requests fehlerfrei verarbeitet werden.

• Atomare Modell-Paketierung

  ---

  Deployment als Einzel-Artefakt: Das gesamte Vorverarbeitungsnetzwerk und das Modell in einer einzigen .joblib-Datei für FastAPI-Inferenz bündeln.

---

1. Die Anatomie von Data Leakage

Data Leakage entsteht, sobald Informationen außerhalb des Trainings-Folds in den Trainingsprozess einfließen. Dies erzeugt eine gefährliche Scheingenauigkeit:

%%{init: {'flowchart': {'wrappingWidth': 350}}}%%
flowchart TB
    subgraph BadPractice ["Fehlerhafter Workflow (Data Leakage)"]
        direction TB
        RAW1["Gesamtdatensatz"]
        LEAK["scaler.fit_transform(Full Dataset)<br/>(Testdaten verzerren Skalierer!)"]
        SPLIT1["train_test_split()<br/>(Testset kontaminiert!)"]
        RAW1 --> LEAK --> SPLIT1
    end

    subgraph Production ["Produktionsmuster (Atomare Pipeline)"]
        direction TB
        RAW2["Gesamtdatensatz"]
        SPLIT2["train_test_split()<br/>(Strikte Trennung Holdout)"]
        PIPE["Pipeline.fit(Train)<br/>(Mediane NUR auf Train!)"]
        TEST["Pipeline.predict(Test)<br/>(Transformation mit Train-Werten)"]
        RAW2 --> SPLIT2 --> PIPE --> TEST
    end

    BadPractice -->|"Atomare Pipeline Loesung"| Production

Die drei Grundregeln zur Vermeidung von Data Leakage:

1. Erst aufteilen, dann vorverarbeiten: Niemals .fit() oder .fit_transform() aufrufen, bevor der Train/Test-Split durchgeführt wurde.
2. Atomare Kapselung: Imputer, Skalierer, Encoder und Estimator müssen zwingend in derselben Pipeline-Instanz liegen.
3. Kreuzvalidierungs-Isolation: cross_val_score oder GridSearchCV muss die unvorverarbeitete Pipeline übergeben werden. Scikit-learn fittet die Vorverarbeitung dann automatisch in jeder Iteration isoliert auf dem jeweiligen Trainings-Fold.

---

2. Heterogene Features verarbeiten: ColumnTransformer

Echte Datensätze kombinieren numerische Messwerte mit kategorialen Strings. Der ColumnTransformer teilt Merkmale in parallele Zweige auf und führt die Ergebnisse in einer einheitlichen Matrix zusammen:

%%{init: {'flowchart': {'wrappingWidth': 350}}}%%
flowchart TB
    INPUT["Rohe Feature-Matrix<br/>(Heterogene Spalten)"]

    subgraph Preprocessing ["ColumnTransformer Pipeline"]
        direction TB
        subgraph NumBranch ["Numerische Features"]
            direction TB
            NUM["Numerische Spalten"]
            NUM_IMP["SimpleImputer(median)"]
            NUM_SCALE["RobustScaler()"]
            NUM --> NUM_IMP --> NUM_SCALE
        end
        subgraph CatBranch ["Kategoriale Features"]
            direction TB
            CAT["Kategoriale Spalten"]
            CAT_IMP["SimpleImputer(mode)"]
            CAT_ENC["OneHotEncoder()"]
            CAT --> CAT_IMP --> CAT_ENC
        end
        NumBranch --> CatBranch
    end

    CONCAT["Einheitliche Feature-Matrix<br/>(Zusammengefuehrte Spalten)"]
    EST["Estimator-Stufe<br/>(Klassifikator / Regressor)"]

    INPUT --> Preprocessing
    Preprocessing --> CONCAT
    CONCAT --> EST

Warum RobustScaler statt StandardScaler?

• StandardScaler: Zieht den Mittelwert (μ) ab und teilt durch die Standardabweichung (σ). Ausreißer verzerren beide Kennzahlen massiv.
• RobustScaler: Zieht den Median (Q₂) ab und skaliert über den Interquartilsabstand (IQR = Q₃ - Q₁). Extreme Infrastruktur-Spitzen verzerren die Skalierungsbasis nicht.

---

3. Praxis-Lab: Ausfall-Vorhersage-Pipeline im Produktivbetrieb

Führen Sie dieses Skript auf Ihrem Debian-System aus, um einen Ausfall-Klassifikator sauber zu trainieren, per Kreuzvalidierung zu prüfen und zu serialisieren:

#!/usr/bin/env python3
# Datei: /opt/ds-ops/train_outage_pipeline.py
import joblib
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.impute import SimpleImputer
from sklearn.metrics import classification_report, roc_auc_score
from sklearn.model_selection import StratifiedKFold, cross_val_score, train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, RobustScaler


def generate_infrastructure_dataset(n_samples: int = 50_000) -> pd.DataFrame:
    """Erzeugt heterogene Telemetriedaten mit realistischen Fehlwerten und Latenzen."""
    rng = np.random.default_rng(seed=2026)

    cpu_pct = rng.uniform(5.0, 98.0, size=n_samples)
    latency_ms = rng.exponential(scale=20.0, size=n_samples)
    regions = rng.choice(["eu-west-1", "eu-central-1", "us-east-1"], size=n_samples)
    tier = rng.choice(["database", "api-gateway", "worker"], size=n_samples)

    # Künstliche Ausfallwahrscheinlichkeit berechnen
    risk_score = (cpu_pct * 0.05) + (latency_ms * 0.08) + (regions == "us-east-1") * 1.5
    prob_failure = 1.0 / (1.0 + np.exp(-(risk_score - 4.5)))
    outage_occurred = (rng.uniform(0, 1, size=n_samples) < prob_failure).astype(int)

    # Realistische NaN-Werte einstreuen
    cpu_pct[rng.choice(n_samples, size=500, replace=False)] = np.nan
    regions[rng.choice(n_samples, size=300, replace=False)] = None

    return pd.DataFrame({
        "cpu_pct": cpu_pct,
        "latency_ms": latency_ms,
        "region": regions,
        "tier": tier,
        "outage_occurred": outage_occurred
    })


def build_zero_leakage_pipeline() -> Pipeline:
    numeric_features = ["cpu_pct", "latency_ms"]
    categorical_features = ["region", "tier"]

    # 1. Numerischer Zweig: Median-Imputation + Robustes Skalieren
    numeric_transformer = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="median")),
        ("scaler", RobustScaler())
    ])

    # 2. Kategorialer Zweig: Modus-Imputation + OneHot-Encoding
    categorical_transformer = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="most_frequent")),
        ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False))
    ])

    # 3. ColumnTransformer zusammensetzen
    preprocessor = ColumnTransformer(
        transformers=[
            ("num", numeric_transformer, numeric_features),
            ("cat", categorical_transformer, categorical_features)
        ],
        remainder="drop"
    )

    # 4. Gesamte Pipeline mit Estimator
    return Pipeline(steps=[
        ("preprocessor", preprocessor),
        ("classifier", HistGradientBoostingClassifier(random_state=42, max_iter=150))
    ])


def main():
    print("1. Generiere Infrastruktur-Telemetriedaten...")
    df = generate_infrastructure_dataset()
    X = df.drop(columns=["outage_occurred"])
    y = df["outage_occurred"]

    # 2. Train-Test Split (Strikt vor jeder Vorverarbeitung!)
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.20, random_state=42, stratify=y
    )

    pipeline = build_zero_leakage_pipeline()

    # 3. Stratifizierte 5-Fold Kreuzvalidierung auf Trainingsdaten
    cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
    cv_scores = cross_val_score(pipeline, X_train, y_train, cv=cv, scoring="roc_auc", n_jobs=-1)
    print(f"5-Fold CV ROC-AUC: {cv_scores.mean():.4f} (± {cv_scores.std():.4f})")

    # 4. Auf Trainingsdaten fitten und auf Testset evaluieren
    pipeline.fit(X_train, y_train)
    y_pred = pipeline.predict(X_test)
    y_prob = pipeline.predict_proba(X_test)[:, 1]

    print("\nHoldout-Test-Evaluierung:")
    print(f"Test ROC-AUC: {roc_auc_score(y_test, y_prob):.4f}")
    print("\nKlassifikationsbericht:\n", classification_report(y_test, y_pred))

    # 5. Atomare Pipeline als Artefakt speichern
    model_path = "/var/models/outage_detector_pipeline.joblib"
    joblib.dump(pipeline, model_path, compress=3)
    print(f"[OK] Pipeline erfolgreich gespeichert unter: {model_path}")


if __name__ == "__main__":
    main()

Warum diese Architektur in Produktion überzeugt:

• Trifft in einem künftigen HTTP-Request eine neue Region wie ap-southeast-1 ein, ignoriert OneHotEncoder(handle_unknown='ignore') diese sicher und gibt einen Null-Vektor aus, statt die Inferenz-API mit einem 500er-Fehler abbrechen zu lassen.
• Fehlt in eingehenden Payloads der Wert für cpu_pct, füllt der SimpleImputer automatisch den exakt beim Training ermittelten Median ein.