Weather Ingestion Pipeline

This lesson teaches you how weather data actually arrives, gets parsed, and lands in Bronze. By the end, you'll understand the file formats (GRIB2, NetCDF), the tooling that reads them, how to schedule incremental ingestion on Databricks, and the key architectural decisions at this layer.

How Weather Data Arrives

Numerical Weather Prediction (NWP) models produce forecast data on fixed schedules. The three main delivery mechanisms:

GRIB2: The Dominant Format

GRIB2 (GRIdded Binary, edition 2) is the standard format for meteorological data. It's compact but not self-describing — you need external tables to decode message meanings.^[1]

What a GRIB2 file contains

• Multiple messages — each is one variable (e.g., wind speed at 100m) for one time step
• Each message has metadata: parameter, level, grid definition, forecast step
• Data is packed as integers with scale/offset — very compact, not directly queryable
• A single download may contain hundreds of messages across variables and time steps

Reading GRIB2 in Python

# The toolchain for GRIB2 → tabular data
import xarray as xr
import cfgrib

# Open GRIB2 file with cfgrib engine
ds = xr.open_dataset(
    "ecmwf_forecast_20260717_00z.grib2",
    engine="cfgrib",
    backend_kwargs={
        "filter_by_keys": {
            "shortName": ["10u", "10v", "2t", "ssrd"],  # wind u/v, temp, irradiance
            "typeOfLevel": "heightAboveGround"
        }
    }
)

# Result: xarray Dataset with dimensions (time, latitude, longitude)
# Variables: u10, v10, t2m, ssrd — each is a 3D array
print(ds)
# <xarray.Dataset>
# Dimensions:  (time: 81, latitude: 721, longitude: 1440)
# Coordinates:
#   * time       (time) datetime64[ns]
#   * latitude   (latitude) float64
#   * longitude  (longitude) float64
# Data variables:
#     u10        (time, latitude, longitude) float32
#     v10        (time, latitude, longitude) float32
#     t2m        (time, latitude, longitude) float32
#     ssrd       (time, latitude, longitude) float32

From xarray to Spark DataFrame

The bridge between meteorology (xarray) and Databricks (Spark) is the key engineering challenge. Here's the pattern:

import xarray as xr
import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import *

spark = SparkSession.builder.getOrCreate()

def grib_to_bronze_df(grib_path: str, init_time: str, source_model: str):
    """Parse a GRIB2 file and return a Spark DataFrame for Bronze."""
    
    # 1. Open with xarray/cfgrib
    ds = xr.open_dataset(grib_path, engine="cfgrib")
    
    # 2. Convert to pandas (flatten grid → rows)
    #    Each row = one grid point × one time step
    df_pd = ds.to_dataframe().reset_index()
    
    # 3. Rename to domain terms
    df_pd = df_pd.rename(columns={
        "latitude": "grid_lat",
        "longitude": "grid_lon",
        "time": "valid_time",  # forecast valid time
        "u10": "wind_u_ms",
        "v10": "wind_v_ms",
        "t2m": "temperature_k",
        "ssrd": "irradiance_jm2",
    })
    
    # 4. Add metadata columns
    df_pd["init_time"] = pd.Timestamp(init_time)
    df_pd["source_model"] = source_model
    df_pd["ingested_at"] = pd.Timestamp.utcnow()
    
    # 5. Convert to Spark DataFrame
    return spark.createDataFrame(df_pd)


# Usage in a Databricks notebook or job
bronze_df = grib_to_bronze_df(
    "/mnt/raw/ecmwf/20260717_00z.grib2",
    init_time="2026-07-17T00:00:00Z",
    source_model="ecmwf_ifs"
)

# Write to Bronze — append-only, partitioned by date for manageability
bronze_df.write.format("delta") \
    .mode("append") \
    .partitionBy("init_time") \
    .saveAsTable("bronze.weather_raw")

Why the Bronze table keeps grid coordinates

Bronze is raw and immutable. It stores grid_lat / grid_lon — not asset_id. The geo-spatial join to assets happens in the Bronze → Silver transformation (covered in Lesson 2). This means if you add new assets later, you can reprocess Bronze without re-downloading.

Bronze Schema

-- Bronze layer: raw weather grid data as-is from source
CREATE TABLE bronze.weather_raw (
  grid_lat        DOUBLE NOT NULL,
  grid_lon        DOUBLE NOT NULL,
  valid_time      TIMESTAMP NOT NULL,     -- what time the forecast is FOR
  init_time       TIMESTAMP NOT NULL,     -- when the model run started
  wind_u_ms       DOUBLE,                 -- eastward wind component
  wind_v_ms       DOUBLE,                 -- northward wind component
  temperature_k   DOUBLE,                 -- 2m temperature in Kelvin
  irradiance_jm2  DOUBLE,                 -- surface solar radiation
  pressure_pa     DOUBLE,
  source_model    STRING NOT NULL,        -- "ecmwf_ifs", "gfs", "icon"
  ingested_at     TIMESTAMP NOT NULL      -- pipeline processing timestamp
)
USING DELTA
PARTITIONED BY (init_time)
TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact'   = 'true',
  'delta.logRetentionDuration'       = '30 days'
);

Orchestrating the Pipeline

Weather ingestion runs on a fixed schedule tied to model availability. Here's how it maps to Databricks Workflows:

Idempotency — The Non-Negotiable

Network failures, API timeouts, and duplicate triggers happen. Your ingestion must be idempotent: running the same job twice for the same init_time must not create duplicate data.

# Pattern: MERGE to enforce idempotency at Bronze
# If init_time + grid_lat + grid_lon + valid_time already exists, skip it

bronze_df.createOrReplaceTempView("new_weather")

spark.sql("""
  MERGE INTO bronze.weather_raw AS target
  USING new_weather AS source
  ON  target.init_time   = source.init_time
  AND target.grid_lat    = source.grid_lat
  AND target.grid_lon    = source.grid_lon
  AND target.valid_time  = source.valid_time
  AND target.source_model = source.source_model
  WHEN NOT MATCHED THEN INSERT *
""")

Scaling Considerations

How big is weather data? Let's estimate for a typical ECMWF forecast:

# Subset to relevant area during parse (e.g., North Sea + Northern Europe)
ds = xr.open_dataset("forecast.grib2", engine="cfgrib")
ds_subset = ds.sel(
    latitude=slice(62, 48),    # North Sea region
    longitude=slice(-5, 15)
)
# Reduces ~1M grid points to ~5,000 — manageable on a single node

Handling Multiple Weather Models

Your team likely consumes multiple NWP sources (ECMWF for accuracy, GFS as backup, ICON for Central European detail). The source_model column in Bronze handles this naturally:

-- Query: compare forecasts from different models for same valid_time
SELECT
  source_model,
  AVG(wind_u_ms) as mean_wind_u,
  AVG(temperature_k) as mean_temp
FROM bronze.weather_raw
WHERE init_time = '2026-07-17T00:00:00Z'
  AND valid_time = '2026-07-18T12:00:00Z'
  AND grid_lat BETWEEN 54.0 AND 56.0
  AND grid_lon BETWEEN 5.0 AND 8.0
GROUP BY source_model

At the Silver layer, you choose which model feeds the production forecast (usually the most recent available, or an ensemble blend). This decision is a Silver-layer concern — Bronze stores all models equally.

Cluster Configuration for Ingestion

Monitoring & Alerting

Weather ingestion has strict timeliness requirements — stale forecasts degrade your power predictions. Build these checks into your pipeline:

• Freshness check: After each run, assert that MAX(init_time) in Bronze is ≤ 6 hours old. Alert if breached.
• Completeness check: Expected row count per model run (known from grid size × time steps). Flag if actual count deviates > 5%.
• Source availability: If the poll in Task 1 finds no new data for > 8 hours, trigger a warning — the upstream source may be delayed or down.
• Schema drift: New variables appearing in GRIB2 files (model updates). Use Delta's schema evolution (mergeSchema) to handle gracefully.

# Example: freshness assertion in a Databricks notebook task
from datetime import datetime, timedelta, timezone

latest = spark.sql("""
  SELECT MAX(init_time) as latest_init FROM bronze.weather_raw
  WHERE source_model = 'ecmwf_ifs'
""").collect()[0]["latest_init"]

threshold = datetime.now(timezone.utc) - timedelta(hours=8)
assert latest >= threshold, f"STALE DATA: latest init_time is {latest}, expected >= {threshold}"

Check Your Understanding

Primary Source

What's Next

You now understand how weather data enters Databricks. The next lesson will cover the Bronze → Silver transformation in depth — the geo-spatial join, wind speed calculation from u/v components, unit conversions, and how to handle multiple forecast versions (init times) for the same valid time.

Questions about GRIB2 parsing, scheduling, or cluster sizing? Ask your teacher — I can dive deeper into cfgrib configuration, multi-model ensemble strategies, or cost estimation for your fleet size.

Rechtliches: Impressum