English version

ML-Pipeline mit MLflow

Diese Lektion behandelt den ML-Lebenszyklus für die Leistungsprognose: Modelle trainieren, Experimente nachverfolgen, Versionen in der Model Registry verwalten und das Champion-Modell für die Batch-Inferenz ausrollen. Am Ende weißt du, wie MLflow in deine Architektur passt und welche Entscheidungen du als Tech Lead triffst.

MLflow in Databricks — das Betriebssystem für ML

MLflow ist direkt in Databricks integriert und deckt vier Aufgabenbereiche ab:^[1]

Ein Windleistungsmodell trainieren

So sieht das Muster für ein Training mit MLflow Tracking und Feature-Store-Anbindung aus:

import mlflow
from mlflow.models import infer_signature
from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup
import lightgbm as lgb
from sklearn.metrics import mean_absolute_error, mean_squared_error
import numpy as np

# 1. Experiment einrichten
mlflow.set_experiment("/Shared/wind-power-prediction")
fe = FeatureEngineeringClient()

# 2. Trainingsset aus dem Feature Store erzeugen (point-in-time korrekt)
labels_df = spark.sql("""
    SELECT asset_id, timestamp, power_kw as actual_power_kw
    FROM prod.silver.power_ts
    WHERE timestamp BETWEEN '2025-01-01' AND '2026-06-30'
      AND power_kw IS NOT NULL
""")

training_set = fe.create_training_set(
    df=labels_df,
    feature_lookups=[
        FeatureLookup(
            table_name="prod.features.wind_power_features",
            lookup_key=["asset_id", "timestamp"]
        )
    ],
    label="actual_power_kw"
)

df = training_set.load_df().toPandas()

# 3. Split (zeitlich — bei Zeitreihen niemals zufällig!)
split_date = "2026-04-01"
train = df[df["timestamp"] < split_date]
test = df[df["timestamp"] >= split_date]

feature_cols = [c for c in train.columns 
                if c not in ["actual_power_kw", "asset_id", "timestamp"]]
X_train, y_train = train[feature_cols], train["actual_power_kw"]
X_test, y_test = test[feature_cols], test["actual_power_kw"]

# 4. Training mit MLflow Tracking
with mlflow.start_run(run_name="lgbm_v1_all_features") as run:
    params = {
        "n_estimators": 500,
        "max_depth": 8,
        "learning_rate": 0.05,
        "num_leaves": 63,
        "min_child_samples": 50,
        "subsample": 0.8,
    }
    mlflow.log_params(params)
    mlflow.log_param("feature_count", len(feature_cols))
    mlflow.log_param("train_size", len(train))
    mlflow.log_param("test_size", len(test))
    mlflow.log_param("split_date", split_date)

    model = lgb.LGBMRegressor(**params)
    model.fit(X_train, y_train)

    # Bewerten
    y_pred = model.predict(X_test)
    mae = mean_absolute_error(y_test, y_pred)
    rmse = np.sqrt(mean_squared_error(y_test, y_pred))
    nmae = mae / df["actual_power_kw"].mean()  # normalisierter MAE

    mlflow.log_metrics({
        "mae_kw": mae,
        "rmse_kw": rmse,
        "nmae": nmae,
        "r2": model.score(X_test, y_test),
    })

    # Modell mit Feature Store loggen (ermöglicht automatischen Feature-Lookup bei der Inferenz)
    fe.log_model(
        model=model,
        artifact_path="model",
        flavor=mlflow.lightgbm,
        training_set=training_set,
        registered_model_name="prod.ml.wind_power_model"
    )

Die Model Registry

Jeder Aufruf von log_model mit einem registered_model_name erzeugt eine neue Version in der Registry. Die Registry führt alle Versionen und unterstützt Aliase für das Deployment:

from mlflow import MlflowClient

client = MlflowClient()

# Eine Version über Aliase in die Produktion befördern
client.set_registered_model_alias(
    name="prod.ml.wind_power_model",
    alias="champion",
    version=3  # die Version, die die Validierung bestanden hat
)

# Einen Challenger für A/B-Tests oder Schattenbetrieb setzen
client.set_registered_model_alias(
    name="prod.ml.wind_power_model",
    alias="challenger",
    version=4
)

Batch-Inferenz-Pipeline

Bei Wetter-zu-Strom läuft die Batch-Inferenz stündlich (nachdem frische Wetterdaten eingetroffen sind). Sie lädt das Champion-Modell, holt die Features aus dem Feature Store und schreibt die Prognosen nach Gold:

import mlflow
from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# 1. Champion-Modell laden
model_uri = "models:/prod.ml.wind_power_model@champion"

# 2. Scoring-Daten vorbereiten (Asset × zukünftige Zeitstempel)
scoring_df = spark.sql("""
    SELECT DISTINCT asset_id, timestamp
    FROM prod.features.wind_power_features
    WHERE timestamp > current_timestamp()
      AND timestamp <= current_timestamp() + INTERVAL 72 HOURS
""")

# 3. Scoring über den Feature Store (automatischer Feature-Lookup!)
predictions = fe.score_batch(
    model_uri=model_uri,
    df=scoring_df
)
# Ergebnis enthält: asset_id, timestamp, prediction (power_kw)

# 4. Metadaten ergänzen und nach Gold schreiben
gold_df = predictions.select(
    "asset_id",
    "timestamp",
    F.col("prediction").alias("forecast_power_kw"),
    F.current_timestamp().alias("produced_at"),
    F.lit("prod.ml.wind_power_model@champion").alias("model_version")
)

gold_df.write.format("delta") \
    .mode("append") \
    .saveAsTable("prod.gold.forecast_ts")

Validierungs-Gate für Modelle

Bevor du eine neue Modellversion zum Champion machst, validierst du sie gegen den aktuellen Champion:

def validate_challenger(challenger_version: int, threshold_mae_improvement: float = 0.05):
    """
    Compare challenger model against current champion on recent data.
    Returns True if challenger should be promoted.
    """
    # Aktuelle Ist-Werte mit beiden Modellen scoren
    recent = spark.sql("""
        SELECT asset_id, timestamp, power_kw as actual
        FROM prod.silver.power_ts
        WHERE timestamp > current_timestamp() - INTERVAL 7 DAYS
          AND power_kw IS NOT NULL
    """)
    
    champion_preds = fe.score_batch(
        model_uri="models:/prod.ml.wind_power_model@champion",
        df=recent.select("asset_id", "timestamp")
    )
    
    challenger_preds = fe.score_batch(
        model_uri=f"models:/prod.ml.wind_power_model/{challenger_version}",
        df=recent.select("asset_id", "timestamp")
    )
    
    # MAE vergleichen
    champion_mae = compute_mae(recent, champion_preds)
    challenger_mae = compute_mae(recent, challenger_preds)
    
    improvement = (champion_mae - challenger_mae) / champion_mae
    
    mlflow.log_metrics({
        "validation_champion_mae": champion_mae,
        "validation_challenger_mae": challenger_mae,
        "validation_improvement_pct": improvement * 100,
    })
    
    return improvement >= threshold_mae_improvement

Modell je Asset oder globales Modell?

Eine Architekturentscheidung, die dir begegnen wird: ein Modell für alle Assets oder je ein Modell pro Asset bzw. Park?

Prüfe dein Verständnis

Primärquelle

Wie geht es weiter

Das Modell liefert Prognosen. Die nächste Lektion behandelt den Gold Layer und die Bereitstellung nach außen — wie du die Prognoseausgabe strukturierst, Konfidenzintervalle berechnest, von Turbine auf Park aggregierst und Ergebnisse an Handelssysteme lieferst.

Fragen zur Modellauswahl (LightGBM vs. neuronale Netze für Wind), zu Ensemble-Strategien oder zur Retraining-Frequenz? Frag deinen Lehrer.

Rechtliches: Impressum