Feature Engineering & Feature Store

This lesson covers how Silver time series become ML-ready features. By the end, you'll understand which features matter for wind power prediction, how the Databricks Feature Store works, and how to design feature pipelines that serve both training and inference consistently.

From Time Series to Features

ML models don't consume raw time series directly. They need features — derived, structured inputs that encode temporal patterns, context, and domain knowledge. The Silver layer gives you clean time series; feature engineering transforms them into what the model actually trains on.

Key Features for Wind Power Prediction

Feature Computation in PySpark

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

def compute_features(weather_ts, power_ts, dim_assets):
    """
    Compute ML features from Silver time series.
    Returns a feature table: one row per (asset_id, timestamp).
    """
    # Window for lagged and rolling features, per asset ordered by time
    w = Window.partitionBy("asset_id").orderBy("timestamp")
    w_6h = w.rowsBetween(-6, -1)   # 6 rows back (hourly data = 6 hours)
    w_24h = w.rowsBetween(-24, -1)

    # Start with weather time series (latest forecast version)
    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")) \
        \
        # Lagged weather
        .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)) \
        \
        # Rolling statistics
        .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)) \
        \
        # Wind ramp: change over last 3 hours
        .withColumn("wind_ramp_3h",
            F.col("wind_speed_hub_ms") - F.lag("wind_speed_hub_ms", 3).over(w)
        ) \
        \
        # Forecast metadata
        .withColumn("lead_time_hours",
            (F.col("timestamp").cast("long") - F.col("forecast_ref_time").cast("long")) / 3600
        )

    # Join asset context (static features)
    features = features.join(
        dim_assets.select(
            "asset_id", "capacity_kw", "hub_height_m",
            "parent_asset_id", "asset_type"
        ),
        on="asset_id",
        how="left"
    )

    # Join recent actual power (for lagged actuals — training only)
    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

The Databricks Feature Store

The Feature Store is a managed layer in Unity Catalog that solves training-serving skew. Features are stored as Delta tables with special metadata that enables automatic lookup during inference.^[1]

Creating a Feature Table

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()

# Create (or update) the feature table in Unity Catalog
fe.create_table(
    name="prod.features.wind_power_features",
    primary_keys=["asset_id", "timestamp"],
    timestamp_keys=["timestamp"],  # enables point-in-time lookups
    df=features_df,
    description="ML features for wind power prediction. Updated hourly."
)

# Subsequent writes: use write_table to append/merge new features
fe.write_table(
    name="prod.features.wind_power_features",
    df=new_features_df,
    mode="merge"  # upsert on primary keys
)

Point-in-time correctness

When training on historical data, you need features as they were at prediction time — not current values. The timestamp_keys parameter enables this:

# Training: look up features as-of each training example's timestamp
training_set = fe.create_training_set(
    df=labels_df,  # has columns: asset_id, timestamp, actual_power_kw
    feature_lookups=[
        FeatureLookup(
            table_name="prod.features.wind_power_features",
            lookup_key=["asset_id", "timestamp"],
            # Point-in-time: gets features valid AT that timestamp
            # Not future-leaking features from after the prediction time
        )
    ],
    label="actual_power_kw"
)

training_df = training_set.load_df()

Feature Pipeline Architecture

Where does feature computation fit in your Workflow?

# In your Databricks Workflow:
# Task 1: Ingest weather → Bronze
# Task 2: Bronze → Silver (geo-join, derive)
# Task 3: Silver → Features  ← THIS LESSON
# Task 4: Features → ML inference → Gold

# Task 3 runs after Silver is updated:
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")
)

# Write to Feature Store
fe.write_table(
    name="prod.features.wind_power_features",
    df=features_df,
    mode="merge"
)

Feature Freshness and the Dual-Access Pattern

Features serve two access patterns (like Silver):

Feature Monitoring

Features drift over time (seasons change, turbines age). Monitor for:

• Distribution drift: wind_speed_mean_6h shifting compared to training data distribution
• Null rate: lagged features are null for the first N rows after gaps — monitor null percentage
• Staleness: if the feature pipeline fails, inference uses stale features — check max(timestamp)
• Correlation breakdown: if wind_speed stops correlating with power_output, something is wrong upstream

# Simple drift check: compare current feature stats to training 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()

Check Your Understanding

Primary Source

What's Next

Features are ready. The next lesson covers the ML pipeline with MLflow — training power prediction models, experiment tracking, model registry, and deploying the champion model for batch inference.

Questions about feature selection for wind vs. solar, handling missing data in lagged features, or feature store performance? Ask your teacher.

Rechtliches: Impressum