English version

Bronze → Silver: Die geo-räumliche Transformation

Diese Lektion behandelt die architektonisch wichtigste Transformation deiner Pipeline: aus rohen Wetterdaten auf einem Raster werden anlagenbezogene Zeitreihen. Am Ende verstehst du den geo-räumlichen Join, die Berechnung abgeleiteter Größen, die Versionierung von Prognosen und wie du das Ganze als verlässliche, inkrementelle Silver-Pipeline aufbaust.

Was der Silver Layer leisten muss

Der Schritt von Bronze nach Silver löst vier Probleme auf einmal:

1. Räumliche Zuordnung: Rasterkoordinaten (lat/lon) → Asset-IDs
2. Ableitung von Größen: Rohkomponenten (Wind u/v) → nutzbare Werte (Windgeschwindigkeit, Windrichtung)
3. Einheiten-Normalisierung: Kelvin → Celsius, J/m² → W/m² usw.
4. Prognose-Versionierung: Mehrere init_times prognostizieren dieselbe valid_time — welche gewinnt?

Nach Silver sieht jeder Konsument (ML-Pipeline, Analystin, nachgelagerte Anwendung) saubere, anlagenbezogene und korrekt typisierte Zeitreihen. Mit Rastergeometrie oder Roheinheiten kommt niemand mehr in Berührung.

Schritt 1: Der geo-räumliche Join

Wetterraster und Anlagen leben in unterschiedlichen Koordinatensystemen. Der Join ordnet jede Anlage der nächstgelegenen Rasterzelle zu — oder interpoliert zwischen mehreren Zellen.

Nächster Nachbar (der einfachste Ansatz)

-- Jede Anlage mit dem nächstgelegenen Rasterpunkt verknüpfen
-- Funktioniert gut, wenn die Rasterauflösung (0,25° ≈ 28 km) gröber ist als der Anlagenabstand
CREATE OR REPLACE TABLE silver.weather_ts AS
SELECT
  a.asset_id,
  w.valid_time         AS timestamp,
  w.init_time          AS forecast_ref_time,
  w.wind_u_ms,
  w.wind_v_ms,
  w.temperature_k,
  w.irradiance_jm2,
  w.pressure_pa,
  w.source_model
FROM bronze.weather_raw w
JOIN silver.dim_assets a
  ON ROUND(w.grid_lat / 0.25) * 0.25 = ROUND(a.latitude / 0.25) * 0.25
  AND ROUND(w.grid_lon / 0.25) * 0.25 = ROUND(a.longitude / 0.25) * 0.25
WHERE w.init_time = '2026-07-17T00:00:00Z'  -- immer nur einen Lauf verarbeiten

Bilineare Interpolation (höhere Genauigkeit)

Wenn Anlagenstandorte zwischen Rasterzellen liegen und Genauigkeit zählt (z. B. bei Offshore-Windparks mit starken Windgradienten), interpolierst du aus den 4 umgebenden Rasterpunkten:

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

def bilinear_interpolate(weather_df, assets_df, grid_res=0.25):
    """
    Für jede Anlage die 4 umgebenden Rasterzellen finden und interpolieren.
    Liefert eine Zeile pro Anlage × valid_time mit interpolierten Wetterwerten.
    """
    # Umschließende Rasterecken für jede Anlage berechnen
    assets_corners = assets_df.select(
        "asset_id", "latitude", "longitude",
        # Auf die Rasterauflösung ab- und aufrunden
        (F.floor(F.col("latitude") / grid_res) * grid_res).alias("lat_lo"),
        (F.ceil(F.col("latitude") / grid_res) * grid_res).alias("lat_hi"),
        (F.floor(F.col("longitude") / grid_res) * grid_res).alias("lon_lo"),
        (F.ceil(F.col("longitude") / grid_res) * grid_res).alias("lon_hi"),
    )

    # Wetterdaten joinen, um alle 4 Ecken zu erhalten
    corners = assets_corners.join(
        weather_df,
        on=(
            (weather_df.grid_lat.isin(
                assets_corners.lat_lo, assets_corners.lat_hi)) &
            (weather_df.grid_lon.isin(
                assets_corners.lon_lo, assets_corners.lon_hi))
        ),
        how="inner"
    )

    # Gewichte anhand der Distanz zu jeder Ecke berechnen
    corners = corners.withColumn(
        "w_lat", 
        1.0 - F.abs(F.col("latitude") - F.col("grid_lat")) / grid_res
    ).withColumn(
        "w_lon",
        1.0 - F.abs(F.col("longitude") - F.col("grid_lon")) / grid_res
    ).withColumn(
        "weight", F.col("w_lat") * F.col("w_lon")
    )

    # Gewichteter Mittelwert je Anlage × valid_time
    return corners.groupBy("asset_id", "valid_time", "init_time", "source_model") \
        .agg(
            F.sum(F.col("wind_u_ms") * F.col("weight")).alias("wind_u_ms"),
            F.sum(F.col("wind_v_ms") * F.col("weight")).alias("wind_v_ms"),
            F.sum(F.col("temperature_k") * F.col("weight")).alias("temperature_k"),
            F.sum(F.col("irradiance_jm2") * F.col("weight")).alias("irradiance_jm2"),
            F.sum(F.col("pressure_pa") * F.col("weight")).alias("pressure_pa"),
        )

Schritt 2: Ableitung der Größen

Wettermodelle liefern rohe physikalische Größen. Deine ML-Modelle und deine Analystinnen brauchen abgeleitete, aussagekräftige Variablen.

from pyspark.sql import functions as F
import math

def derive_weather_variables(df, hub_height_m=100.0, roughness_alpha=0.14):
    """
    Rohe Wetterkomponenten in nutzbare abgeleitete Größen umwandeln.
    Setzt voraus, dass df enthält: wind_u_ms, wind_v_ms, temperature_k, irradiance_jm2
    """
    return df.withColumn(
        "wind_speed_ms",
        F.sqrt(F.col("wind_u_ms")**2 + F.col("wind_v_ms")**2)
    ).withColumn(
        "wind_direction_deg",
        (F.degrees(F.atan2(-F.col("wind_u_ms"), -F.col("wind_v_ms"))) + 360) % 360
    ).withColumn(
        "temperature_c",
        F.col("temperature_k") - 273.15
    ).withColumn(
        # Extrapolation auf Nabenhöhe über das Potenzgesetz
        # α = 0.14 ist typisch für offenes Gelände / Offshore
        "wind_speed_hub_ms",
        F.col("wind_speed_ms") * F.pow(F.lit(hub_height_m / 10.0), F.lit(roughness_alpha))
    ).drop("wind_u_ms", "wind_v_ms", "temperature_k")

Schritt 3: Einheiten-Normalisierung

Silver ist der eine Ort, an dem Einheiten vereinheitlicht werden. Dokumentiere sie klar:

-- Konventionen im Silver Layer (in den Unity-Catalog-Tabellenkommentaren dokumentieren)
-- wind_speed_ms:       m/s (auf 10 m Referenzhöhe)
-- wind_speed_hub_ms:   m/s (auf die Nabenhöhe der Anlage extrapoliert)
-- wind_direction_deg:  Grad von Nord, im Uhrzeigersinn (meteorologische Konvention)
-- temperature_c:       Celsius
-- irradiance_wm2:      W/m² (momentane Leistung, keine kumulierte Energie)
-- pressure_hpa:        Hektopascal (= Millibar)
-- timestamp:           UTC, kein Zeitzonen-Offset gespeichert

Schritt 4: Prognose-Versionierung

Mehrere Modellläufe sagen denselben Zeitpunkt in der Zukunft voraus. Der 00Z-Lauf prognostiziert den morgigen Mittag — und der 06Z-Lauf ebenfalls, nur mit frischeren Daten. Welchen soll Silver ausliefern?

Strategie: alle Versionen behalten, die neueste markieren

-- Die Silver-Tabelle speichert ALLE Prognoseversionen
-- Eine berechnete Spalte oder ein View markiert die "aktuellste verfügbare" je valid_time

CREATE OR REPLACE VIEW silver.weather_ts_latest AS
WITH ranked AS (
  SELECT *,
    ROW_NUMBER() OVER (
      PARTITION BY asset_id, timestamp, source_model
      ORDER BY forecast_ref_time DESC  -- der jüngste Init gewinnt
    ) AS version_rank
  FROM silver.weather_ts
)
SELECT * FROM ranked WHERE version_rank = 1;

Die vollständige Silver-Pipeline

Alles zusammengesetzt als ein einziger inkrementeller Job:

from pyspark.sql import functions as F

def bronze_to_silver(spark, init_time: str, source_model: str):
    """
    Einen Modelllauf von Bronze nach Silver verarbeiten.
    Idempotent: überschreibt die Partition für diese init_time + source_model.
    """
    # 1. Den neuen Bronze-Batch lesen
    bronze = spark.read.table("bronze.weather_raw") \
        .filter(F.col("init_time") == init_time) \
        .filter(F.col("source_model") == source_model)

    # 2. Asset-Dimension laden (kleine Tabelle — Broadcast Join)
    assets = spark.read.table("silver.dim_assets") \
        .filter(F.col("asset_type").isin("wind_park", "turbine"))

    # 3. Geo-räumlicher Join (nächstgelegener Rasterpunkt)
    grid_res = 0.25
    assets_snapped = assets.withColumn(
        "snap_lat", F.round(F.col("latitude") / grid_res) * grid_res
    ).withColumn(
        "snap_lon", F.round(F.col("longitude") / grid_res) * grid_res
    )
    
    bronze_snapped = bronze.withColumn(
        "snap_lat", F.round(F.col("grid_lat") / grid_res) * grid_res
    ).withColumn(
        "snap_lon", F.round(F.col("grid_lon") / grid_res) * grid_res
    )

    joined = bronze_snapped.join(
        F.broadcast(assets_snapped.select(
            "asset_id", "snap_lat", "snap_lon", "hub_height_m"
        )),
        on=["snap_lat", "snap_lon"],
        how="inner"
    )

    # 4. Größen ableiten
    silver = joined.withColumn(
        "wind_speed_ms",
        F.sqrt(F.col("wind_u_ms")**2 + F.col("wind_v_ms")**2)
    ).withColumn(
        "wind_direction_deg",
        (F.degrees(F.atan2(-F.col("wind_u_ms"), -F.col("wind_v_ms"))) + 360) % 360
    ).withColumn(
        "temperature_c", F.col("temperature_k") - 273.15
    ).withColumn(
        "wind_speed_hub_ms",
        F.col("wind_speed_ms") * F.pow(
            F.col("hub_height_m") / 10.0, F.lit(0.14)
        )
    ).withColumn(
        "pressure_hpa", F.col("pressure_pa") / 100.0
    )

    # 5. Finales Silver-Schema auswählen
    result = silver.select(
        "asset_id",
        F.col("valid_time").alias("timestamp"),
        F.col("init_time").alias("forecast_ref_time"),
        "wind_speed_ms",
        "wind_speed_hub_ms",
        "wind_direction_deg",
        "temperature_c",
        F.col("irradiance_jm2").alias("irradiance_jm2"),  # Umrechnung nachgelagert
        "pressure_hpa",
        "source_model"
    )

    # 6. Schreiben — MERGE für Idempotenz
    result.createOrReplaceTempView("new_silver")
    spark.sql("""
        MERGE INTO silver.weather_ts AS target
        USING new_silver AS source
        ON  target.asset_id         = source.asset_id
        AND target.timestamp        = source.timestamp
        AND target.forecast_ref_time = source.forecast_ref_time
        AND target.source_model     = source.source_model
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

    return result.count()

Inkrementelle Verarbeitung

Du verarbeitest nicht bei jedem Lauf ganz Bronze neu. Die Pipeline soll nur neue Daten seit dem letzten Lauf transformieren.

# Parametergesteuerter Ansatz (am einfachsten, empfohlen für den Start)
# Der Workflow übergibt init_time als Widget-Parameter
dbutils.widgets.text("init_time", "")
dbutils.widgets.text("source_model", "ecmwf_ifs")

init_time = dbutils.widgets.get("init_time")
source_model = dbutils.widgets.get("source_model")

rows = bronze_to_silver(spark, init_time, source_model)
print(f"Processed {rows} rows for {source_model} @ {init_time}")

Datenqualitätsprüfungen in Silver

Silver ist die Stelle, an der du Datenqualität durchsetzt. Wenn Bronze "alles speichern" heißt, dann heißt Silver "vertrauen, aber prüfen".

# Qualitätserwartungen — der Batch scheitert bei Verletzung
from pyspark.sql import functions as F

def validate_silver_batch(df):
    """Wirft einen Fehler, wenn Qualitätsschwellen überschritten werden."""
    stats = df.agg(
        F.count("*").alias("total_rows"),
        F.sum(F.when(F.col("wind_speed_ms") < 0, 1).otherwise(0)).alias("negative_wind"),
        F.sum(F.when(F.col("wind_speed_ms") > 60, 1).otherwise(0)).alias("extreme_wind"),
        F.sum(F.when(F.col("temperature_c") < -60, 1).otherwise(0)).alias("extreme_cold"),
        F.sum(F.when(F.col("temperature_c") > 60, 1).otherwise(0)).alias("extreme_hot"),
        F.sum(F.when(F.col("wind_speed_ms").isNull(), 1).otherwise(0)).alias("null_wind"),
    ).collect()[0]
    
    total = stats["total_rows"]
    issues = []
    
    if stats["negative_wind"] > 0:
        issues.append(f"Negative wind speed: {stats['negative_wind']} rows")
    if stats["extreme_wind"] > total * 0.001:
        issues.append(f"Extreme wind (>60 m/s): {stats['extreme_wind']} rows")
    if stats["null_wind"] > total * 0.05:
        issues.append(f"Null wind speed: {stats['null_wind']} rows ({stats['null_wind']/total*100:.1f}%)")
    
    if issues:
        raise ValueError(f"Silver quality check FAILED:\n" + "\n".join(issues))
    
    print(f"✓ Quality check passed: {total} rows, no issues")

Verständnis prüfen

Primärquelle

Wie es weitergeht

Nachdem die Wetterdaten jetzt als saubere, anlagenbezogene Zeitreihen in Silver liegen, behandelt die nächste Lektion die Unity-Catalog-Governance — wie du Kataloge, Schemas und Berechtigungen so organisierst, dass dein Team (Data Engineers, Data Scientists, Analystinnen) sicher an diesen Tabellen zusammenarbeiten kann.

Fragen zur Interpolationsgenauigkeit, zur Strategie der Prognose-Versionierung oder zur MERGE-Performance im großen Maßstab? Frag deinen Lehrer — ich kann konkrete Szenarien mit deiner Flottengröße und deinen Datenvolumina durchspielen.

Rechtliches: Impressum