Bronze → Silver: The Geo-Spatial Transform

This lesson covers the most architecturally significant transformation in your pipeline: turning raw gridded weather data into asset-keyed time series. By the end, you'll understand the geo-spatial join, derived variable calculations, forecast versioning, and how to structure this as a reliable, incremental Silver pipeline.

What the Silver Layer Must Accomplish

The Bronze → Silver step resolves four problems at once:

1. Spatial resolution: Grid coordinates (lat/lon) → asset IDs
2. Variable derivation: Raw components (wind u/v) → usable quantities (wind speed, direction)
3. Unit normalisation: Kelvin → Celsius, J/m² → W/m², etc.
4. Forecast versioning: Multiple init_times predict the same valid_time — which one wins?

After Silver, every consumer (ML pipeline, analyst, downstream app) sees clean, asset-keyed, properly-typed time series. They never touch grid geometry or raw units again.

Step 1: The Geo-Spatial Join

Weather grids and assets live in different coordinate systems. The join maps each asset to its nearest grid cell (or interpolates between cells).

Nearest-neighbour (simplest approach)

-- Join each asset to the nearest grid point
-- Works well when grid resolution (0.25° ≈ 28km) is coarser than asset spacing
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'  -- process one run at a time

Bilinear interpolation (higher accuracy)

When asset locations fall between grid cells and accuracy matters (e.g., offshore wind farms where wind gradients are steep), interpolate from the 4 surrounding grid points:

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

def bilinear_interpolate(weather_df, assets_df, grid_res=0.25):
    """
    For each asset, find the 4 surrounding grid cells and interpolate.
    Returns one row per asset × valid_time with interpolated weather values.
    """
    # Calculate bounding grid corners for each asset
    assets_corners = assets_df.select(
        "asset_id", "latitude", "longitude",
        # Floor and ceil to grid resolution
        (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"),
    )

    # Join weather data to get all 4 corners
    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"
    )

    # Compute weights based on distance to each corner
    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")
    )

    # Weighted average per asset × 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"),
        )

Step 2: Variable Derivation

Weather models output raw physical quantities. Your ML models and analysts need derived, meaningful variables.

from pyspark.sql import functions as F
import math

def derive_weather_variables(df, hub_height_m=100.0, roughness_alpha=0.14):
    """
    Transform raw weather components into usable derived variables.
    Assumes df has: 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(
        # Hub-height extrapolation using power law
        # α = 0.14 is typical for open terrain / 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")

Step 3: Unit Normalisation

Silver is the single place where units are standardised. Document them clearly:

-- Silver layer conventions (document in Unity Catalog table comments)
-- wind_speed_ms:       m/s (at 10m reference height)
-- wind_speed_hub_ms:   m/s (extrapolated to asset hub height)
-- wind_direction_deg:  degrees from North, clockwise (meteorological convention)
-- temperature_c:       Celsius
-- irradiance_wm2:      W/m² (instantaneous power, not accumulated energy)
-- pressure_hpa:        hectopascals (= millibars)
-- timestamp:           UTC, no timezone offset stored

Step 4: Forecast Versioning

Multiple model runs predict the same future time. The 00Z run predicts tomorrow noon, and so does the 06Z run (with fresher data). Which one should Silver expose?

Strategy: Keep all versions, flag the latest

-- Silver table stores ALL forecast versions
-- A computed column or view flags the "latest available" for each 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  -- most recent init wins
    ) AS version_rank
  FROM silver.weather_ts
)
SELECT * FROM ranked WHERE version_rank = 1;

The Complete Silver Pipeline

Putting it all together as a single incremental job:

from pyspark.sql import functions as F

def bronze_to_silver(spark, init_time: str, source_model: str):
    """
    Process one model run from Bronze → Silver.
    Idempotent: overwrites the partition for this init_time + source_model.
    """
    # 1. Read the new Bronze batch
    bronze = spark.read.table("bronze.weather_raw") \
        .filter(F.col("init_time") == init_time) \
        .filter(F.col("source_model") == source_model)

    # 2. Load asset dimension (small table — broadcast join)
    assets = spark.read.table("silver.dim_assets") \
        .filter(F.col("asset_type").isin("wind_park", "turbine"))

    # 3. Geo-spatial join (nearest grid point)
    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. Derive variables
    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. Select final Silver schema
    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"),  # convert downstream
        "pressure_hpa",
        "source_model"
    )

    # 6. Write — MERGE for idempotency
    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()

Incremental Processing

You don't reprocess all of Bronze every run. The pipeline should only transform new data since the last run.

# Parameter-driven approach (simplest, recommended to start)
# The Workflow passes init_time as a 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}")

Data Quality Checks at Silver

Silver is where you enforce data quality. If Bronze is "store everything", Silver is "trust but verify."

# Quality expectations — fail the batch if violated
from pyspark.sql import functions as F

def validate_silver_batch(df):
    """Raise if quality thresholds are breached."""
    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")

Check Your Understanding

Primary Source

What's Next

With weather data now in Silver as clean, asset-keyed time series, the next lesson covers Unity Catalog governance — how to organise catalogs, schemas, and permissions so your team (data engineers, data scientists, analysts) can collaborate safely on these tables.

Questions about interpolation accuracy, forecast versioning strategy, or the MERGE performance at scale? Ask your teacher — I can work through specific scenarios with your fleet size and data volumes.

Rechtliches: Impressum