English version

Feature Engineering & Feature Store

Diese Lektion zeigt, wie aus Silver-Zeitreihen ML-taugliche Features werden. Am Ende weißt du, welche Features für die Windleistungsprognose zählen, wie der Databricks Feature Store funktioniert und wie du Feature-Pipelines entwirfst, die Training und Inferenz konsistent bedienen.

Von Zeitreihen zu Features

ML-Modelle verarbeiten keine rohen Zeitreihen. Sie brauchen Features — abgeleitete, strukturierte Eingaben, die zeitliche Muster, Kontext und Fachwissen kodieren. Der Silver Layer liefert dir saubere Zeitreihen; das Feature Engineering macht daraus das, worauf das Modell tatsächlich trainiert.

Zentrale Features für die Windleistungsprognose

Feature-Berechnung in PySpark

from pyspark.sql import functions as F
from pyspark.sql.window import Window

def compute_features(weather_ts, power_ts, dim_assets):
    """
    ML-Features aus den Silver-Zeitreihen berechnen.
    Liefert eine Feature-Tabelle: eine Zeile je (asset_id, timestamp).
    """
    # Fenster für verzögerte und gleitende Features, je Anlage nach Zeit sortiert
    w = Window.partitionBy("asset_id").orderBy("timestamp")
    w_6h = w.rowsBetween(-6, -1)   # 6 Zeilen zurück (Stundendaten = 6 Stunden)
    w_24h = w.rowsBetween(-24, -1)

    # Basis: Wetter-Zeitreihe (jeweils neueste Prognoseversion)
    features = weather_ts.filter(F.col("version_rank") == 1) \
        .withColumn("hour_of_day", F.hour("timestamp")) \
        .withColumn("day_of_year", F.dayofyear("timestamp")) \
        .withColumn("month", F.month("timestamp")) \
        \
        # Verzögerte Wetterwerte
        .withColumn("wind_speed_1h_ago", F.lag("wind_speed_hub_ms", 1).over(w)) \
        .withColumn("wind_speed_3h_ago", F.lag("wind_speed_hub_ms", 3).over(w)) \
        .withColumn("wind_speed_6h_ago", F.lag("wind_speed_hub_ms", 6).over(w)) \
        \
        # Gleitende Statistiken
        .withColumn("wind_speed_mean_6h", F.avg("wind_speed_hub_ms").over(w_6h)) \
        .withColumn("wind_speed_std_6h", F.stddev("wind_speed_hub_ms").over(w_6h)) \
        .withColumn("wind_speed_max_24h", F.max("wind_speed_hub_ms").over(w_24h)) \
        \
        # Windrampe: Änderung über die letzten 3 Stunden
        .withColumn("wind_ramp_3h",
            F.col("wind_speed_hub_ms") - F.lag("wind_speed_hub_ms", 3).over(w)
        ) \
        \
        # Prognose-Metadaten
        .withColumn("lead_time_hours",
            (F.col("timestamp").cast("long") - F.col("forecast_ref_time").cast("long")) / 3600
        )

    # Anlagenkontext joinen (statische Features)
    features = features.join(
        dim_assets.select(
            "asset_id", "capacity_kw", "hub_height_m",
            "parent_asset_id", "asset_type"
        ),
        on="asset_id",
        how="left"
    )

    # Jüngste Ist-Leistung joinen (verzögerte Istwerte — nur fürs Training)
    if power_ts is not None:
        power_lagged = power_ts.select(
            "asset_id", "timestamp",
            F.lag("power_kw", 1).over(w).alias("power_1h_ago"),
            F.lag("power_kw", 3).over(w).alias("power_3h_ago"),
        )
        features = features.join(power_lagged, on=["asset_id", "timestamp"], how="left")

    return features

Der Databricks Feature Store

Der Feature Store ist eine verwaltete Schicht im Unity Catalog, die Training-Serving-Skew löst. Features werden als Delta-Tabellen mit speziellen Metadaten abgelegt, die während der Inferenz einen automatischen Lookup ermöglichen.^[1]

Eine Feature-Tabelle anlegen

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# Feature-Tabelle im Unity Catalog anlegen (oder aktualisieren)
fe.create_table(
    name="prod.features.wind_power_features",
    primary_keys=["asset_id", "timestamp"],
    timestamp_keys=["timestamp"],  # ermöglicht Point-in-Time-Lookups
    df=features_df,
    description="ML features for wind power prediction. Updated hourly."
)

# Spätere Schreibvorgänge: write_table zum Anhängen/Mergen neuer Features
fe.write_table(
    name="prod.features.wind_power_features",
    df=new_features_df,
    mode="merge"  # Upsert auf die Primärschlüssel
)

Point-in-Time-Korrektheit

Beim Training auf historischen Daten brauchst du die Features so, wie sie zum Vorhersagezeitpunkt aussahen — nicht die heutigen Werte. Der Parameter timestamp_keys ermöglicht das:

# Training: Features zum Zeitstempel jedes Trainingsbeispiels nachschlagen
training_set = fe.create_training_set(
    df=labels_df,  # enthält die Spalten: asset_id, timestamp, actual_power_kw
    feature_lookups=[
        FeatureLookup(
            table_name="prod.features.wind_power_features",
            lookup_key=["asset_id", "timestamp"],
            # Point-in-Time: liefert die zu diesem Zeitstempel gültigen Features
            # Keine Features aus der Zukunft nach dem Vorhersagezeitpunkt
        )
    ],
    label="actual_power_kw"
)

training_df = training_set.load_df()

Architektur der Feature-Pipeline

Wo genau sitzt die Feature-Berechnung in deinem Workflow?

# In deinem Databricks Workflow:
# Task 1: Wetter ingesten → Bronze
# Task 2: Bronze → Silver (Geo-Join, Ableitungen)
# Task 3: Silver → Features  ← DIESE LEKTION
# Task 4: Features → ML-Inferenz → Gold

# Task 3 läuft, nachdem Silver aktualisiert wurde:
features_df = compute_features(
    weather_ts=spark.read.table("prod.silver.weather_ts_latest"),
    power_ts=spark.read.table("prod.silver.power_ts"),
    dim_assets=spark.read.table("prod.silver.dim_assets")
)

# In den Feature Store schreiben
fe.write_table(
    name="prod.features.wind_power_features",
    df=features_df,
    mode="merge"
)

Feature-Aktualität und das Dual-Access-Muster

Features bedienen zwei Zugriffsmuster (wie schon Silver):

Feature-Monitoring

Features driften mit der Zeit (Jahreszeiten wechseln, Turbinen altern). Überwache:

• Verteilungsdrift: wind_speed_mean_6h verschiebt sich gegenüber der Verteilung der Trainingsdaten
• Null-Rate: verzögerte Features sind für die ersten N Zeilen nach einer Lücke null — überwache den Null-Anteil
• Veraltung: fällt die Feature-Pipeline aus, arbeitet die Inferenz mit veralteten Features — prüfe max(timestamp)
• Zusammenbruch von Korrelationen: korreliert wind_speed nicht mehr mit power_output, stimmt vorgelagert etwas nicht

# Einfache Drift-Prüfung: aktuelle Feature-Statistiken gegen die Trainings-Baseline
from pyspark.sql import functions as F

current_stats = spark.read.table("prod.features.wind_power_features") \
    .filter(F.col("timestamp") > F.current_timestamp() - F.expr("INTERVAL 7 DAYS")) \
    .agg(
        F.mean("wind_speed_hub_ms").alias("mean_wind"),
        F.stddev("wind_speed_hub_ms").alias("std_wind"),
        F.mean("lead_time_hours").alias("mean_lead"),
        (F.sum(F.when(F.col("wind_speed_1h_ago").isNull(), 1)) / F.count("*")).alias("null_rate")
    )
current_stats.show()

Verständnis prüfen

Primärquelle

Wie es weitergeht

Die Features stehen. Die nächste Lektion behandelt die ML-Pipeline mit MLflow — Modelle für die Leistungsprognose trainieren, Experimente nachverfolgen, die Model Registry nutzen und das Champion-Modell für Batch-Inferenz ausrollen.

Fragen zur Feature-Auswahl bei Wind vs. Solar, zum Umgang mit fehlenden Werten in verzögerten Features oder zur Performance des Feature Store? Frag deinen Lehrer.

Rechtliches: Impressum