English version

Wetterdaten-Ingestion-Pipeline

Diese Lektion zeigt dir, wie Wetterdaten tatsächlich ankommen, geparst werden und in Bronze landen. Am Ende verstehst du die Dateiformate (GRIB2, NetCDF), das Werkzeug, das sie liest, wie du inkrementelle Ingestion auf Databricks einplanst, und die zentralen Architekturentscheidungen dieser Schicht.

Wie Wetterdaten ankommen

Numerische Wettervorhersagemodelle (NWP) erzeugen Prognosedaten nach festen Zeitplänen. Die drei wichtigsten Auslieferungswege:

GRIB2: das dominierende Format

GRIB2 (GRIdded Binary, Edition 2) ist das Standardformat für meteorologische Daten. Es ist kompakt, aber nicht selbstbeschreibend — du brauchst externe Tabellen, um die Bedeutung der Messages zu dekodieren.^[1]

Was eine GRIB2-Datei enthält

• Mehrere Messages — jede steht für eine Variable (z. B. Windgeschwindigkeit in 100 m) zu einem Zeitschritt
• Jede Message hat Metadaten: Parameter, Level, Gitterdefinition, Prognoseschritt
• Daten sind als Ganzzahlen mit Skalierung/Offset gepackt — sehr kompakt, aber nicht direkt abfragbar
• Ein einzelner Download kann Hunderte Messages über Variablen und Zeitschritte hinweg enthalten

GRIB2 in Python lesen

# Die Toolchain für GRIB2 → tabellarische Daten
import xarray as xr
import cfgrib

# GRIB2-Datei mit der cfgrib-Engine öffnen
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., Einstrahlung
            "typeOfLevel": "heightAboveGround"
        }
    }
)

# Ergebnis: xarray-Dataset mit den Dimensionen (time, latitude, longitude)
# Variablen: u10, v10, t2m, ssrd — jeweils ein 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

Von xarray zum Spark DataFrame

Die Brücke zwischen Meteorologie (xarray) und Databricks (Spark) ist die zentrale technische Herausforderung. So sieht das Muster aus:

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):
    """GRIB2-Datei parsen und ein Spark DataFrame für Bronze zurückgeben."""
    
    # 1. Mit xarray/cfgrib öffnen
    ds = xr.open_dataset(grib_path, engine="cfgrib")
    
    # 2. Nach pandas konvertieren (Gitter → Zeilen flach ziehen)
    #    Jede Zeile = ein Gitterpunkt × ein Zeitschritt
    df_pd = ds.to_dataframe().reset_index()
    
    # 3. Auf Domänenbegriffe umbenennen
    df_pd = df_pd.rename(columns={
        "latitude": "grid_lat",
        "longitude": "grid_lon",
        "time": "valid_time",  # Gültigkeitszeit der Prognose
        "u10": "wind_u_ms",
        "v10": "wind_v_ms",
        "t2m": "temperature_k",
        "ssrd": "irradiance_jm2",
    })
    
    # 4. Metadaten-Spalten ergänzen
    df_pd["init_time"] = pd.Timestamp(init_time)
    df_pd["source_model"] = source_model
    df_pd["ingested_at"] = pd.Timestamp.utcnow()
    
    # 5. In ein Spark DataFrame umwandeln
    return spark.createDataFrame(df_pd)


# Verwendung in einem Databricks-Notebook oder -Job
bronze_df = grib_to_bronze_df(
    "/mnt/raw/ecmwf/20260717_00z.grib2",
    init_time="2026-07-17T00:00:00Z",
    source_model="ecmwf_ifs"
)

# Nach Bronze schreiben — nur anhängend, zur Handhabbarkeit nach Datum partitioniert
bronze_df.write.format("delta") \
    .mode("append") \
    .partitionBy("init_time") \
    .saveAsTable("bronze.weather_raw")

Warum die Bronze-Tabelle Gitterkoordinaten behält

Bronze ist roh und unveränderlich. Die Tabelle speichert grid_lat / grid_lon — nicht asset_id. Der geo-räumliche Join auf Assets passiert in der Transformation Bronze → Silver (behandelt in Lektion 2). Das heißt: Wenn du später neue Assets hinzufügst, kannst du Bronze neu verarbeiten, ohne die Daten erneut herunterzuladen.

Bronze-Schema

-- Bronze-Schicht: rohe Wettergitterdaten unverändert aus der Quelle
CREATE TABLE bronze.weather_raw (
  grid_lat        DOUBLE NOT NULL,
  grid_lon        DOUBLE NOT NULL,
  valid_time      TIMESTAMP NOT NULL,     -- für welchen Zeitpunkt die Prognose gilt
  init_time       TIMESTAMP NOT NULL,     -- wann der Modelllauf startete
  wind_u_ms       DOUBLE,                 -- Windkomponente nach Osten
  wind_v_ms       DOUBLE,                 -- Windkomponente nach Norden
  temperature_k   DOUBLE,                 -- 2-m-Temperatur in Kelvin
  irradiance_jm2  DOUBLE,                 -- solare Einstrahlung an der Oberfläche
  pressure_pa     DOUBLE,
  source_model    STRING NOT NULL,        -- "ecmwf_ifs", "gfs", "icon"
  ingested_at     TIMESTAMP NOT NULL      -- Verarbeitungszeitpunkt der Pipeline
)
USING DELTA
PARTITIONED BY (init_time)
TBLPROPERTIES (
  'delta.autoOptimize.optimizeWrite' = 'true',
  'delta.autoOptimize.autoCompact'   = 'true',
  'delta.logRetentionDuration'       = '30 days'
);

Die Pipeline orchestrieren

Die Wetter-Ingestion läuft nach einem festen Zeitplan, der an die Verfügbarkeit der Modelle gekoppelt ist. So bildet sich das auf Databricks Workflows ab:

Idempotenz — nicht verhandelbar

Netzwerkausfälle, API-Timeouts und doppelte Trigger passieren. Deine Ingestion muss idempotent sein: Läuft derselbe Job zweimal für dieselbe init_time, dürfen keine doppelten Daten entstehen.

# Muster: MERGE erzwingt Idempotenz in Bronze
# Existiert init_time + grid_lat + grid_lon + valid_time bereits, wird übersprungen

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 *
""")

Überlegungen zur Skalierung

Wie groß sind Wetterdaten? Schätzen wir das für eine typische ECMWF-Prognose ab:

# Beim Parsen auf das relevante Gebiet zuschneiden (z. B. Nordsee + Nordeuropa)
ds = xr.open_dataset("forecast.grib2", engine="cfgrib")
ds_subset = ds.sel(
    latitude=slice(62, 48),    # Region Nordsee
    longitude=slice(-5, 15)
)
# Reduziert ca. 1 Mio. Gitterpunkte auf ca. 5.000 — auf einem Single-Node handhabbar

Mehrere Wettermodelle handhaben

Dein Team nutzt vermutlich mehrere NWP-Quellen (ECMWF für Genauigkeit, GFS als Backup, ICON für mitteleuropäische Details). Die Spalte source_model in Bronze deckt das ganz natürlich ab:

-- Abfrage: Prognosen verschiedener Modelle für dieselbe valid_time vergleichen
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

In der Silver-Schicht entscheidest du, welches Modell die Produktionsprognose speist (meist das aktuellste verfügbare oder eine Ensemble-Mischung). Diese Entscheidung gehört in Silver — Bronze speichert alle Modelle gleichberechtigt.

Cluster-Konfiguration für die Ingestion

Monitoring & Alerting

Die Wetter-Ingestion hat strenge Anforderungen an Aktualität — veraltete Prognosen verschlechtern deine Stromvorhersagen. Baue diese Prüfungen in deine Pipeline ein:

• Aktualitätsprüfung: Stelle nach jedem Lauf sicher, dass MAX(init_time) in Bronze höchstens 6 Stunden alt ist. Alarmiere bei Verletzung.
• Vollständigkeitsprüfung: Erwartete Zeilenzahl je Modelllauf (bekannt aus Gittergröße × Zeitschritten). Melde, wenn die tatsächliche Zahl um > 5 % abweicht.
• Verfügbarkeit der Quelle: Findet die Abfrage in Task 1 länger als 8 Stunden keine neuen Daten, löse eine Warnung aus — die Upstream-Quelle könnte verzögert oder ausgefallen sein.
• Schema-Drift: Neue Variablen tauchen in GRIB2-Dateien auf (Modell-Updates). Nutze die Schema-Evolution von Delta (mergeSchema), um das elegant abzufangen.

# Beispiel: Aktualitätsprüfung in einem 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}"

Prüfe dein Verständnis

Primärquelle

Wie geht es weiter

Du verstehst jetzt, wie Wetterdaten nach Databricks gelangen. Die nächste Lektion behandelt die Transformation Bronze → Silver im Detail — den geo-räumlichen Join, die Berechnung der Windgeschwindigkeit aus u/v-Komponenten, Einheitenumrechnungen und den Umgang mit mehreren Prognoseversionen (Init-Zeiten) für dieselbe Gültigkeitszeit.

Fragen zum GRIB2-Parsing, zu Zeitplänen oder zur Cluster-Dimensionierung? Frag deinen Lehrer — ich kann tiefer in die cfgrib-Konfiguration, Multi-Modell-Ensemble-Strategien oder eine Kostenschätzung für deine Flottengröße einsteigen.

Rechtliches: Impressum