Weather and Climate Analytics¶

# Databricks notebook source
# MAGIC %md
# MAGIC # Storm Severity Analysis
# MAGIC Bronze-to-Silver transformation with severity classification and damage normalization

# COMMAND ----------

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

# Load Bronze storm events
df_storms = spark.read.format("delta").load(
    "abfss://lh_bronze@onelake.dfs.fabric.microsoft.com/Tables/bronze_noaa_storm_events"
)

# --- Normalize Damage Values ---
# NOAA encodes damage as "25K", "1.5M", "500" etc.
def parse_damage_udf(value):
    """Convert NOAA damage notation to numeric."""
    if value is None or value == "":
        return 0.0
    value = str(value).strip().upper()
    multipliers = {"K": 1_000, "M": 1_000_000, "B": 1_000_000_000}
    for suffix, mult in multipliers.items():
        if value.endswith(suffix):
            return float(value[:-1]) * mult
    try:
        return float(value)
    except ValueError:
        return 0.0

parse_damage = F.udf(parse_damage_udf, "double")

df_normalized = df_storms.withColumn(
    "property_damage_amt", parse_damage(F.col("damage_property"))
).withColumn(
    "crop_damage_amt", parse_damage(F.col("damage_crops"))
).withColumn(
    "total_damage_amt",
    F.col("property_damage_amt") + F.col("crop_damage_amt")
).withColumn(
    "total_casualties",
    F.col("deaths_direct") + F.col("deaths_indirect")
    + F.col("injuries_direct") + F.col("injuries_indirect")
)

# --- Severity Classification ---
df_classified = df_normalized.withColumn(
    "severity_tier",
    F.when(
        (F.col("total_damage_amt") > 10_000_000) | (F.col("total_casualties") > 10),
        "CATASTROPHIC"
    ).when(
        (F.col("total_damage_amt") > 1_000_000) | (F.col("total_casualties") > 3),
        "SEVERE"
    ).when(
        (F.col("total_damage_amt") > 100_000) | (F.col("total_casualties") > 0),
        "SIGNIFICANT"
    ).otherwise("MINOR")
)

# --- Decadal Trend Analysis ---
df_trends = df_classified.withColumn(
    "event_year", F.year("begin_date")
).withColumn(
    "decade",
    (F.floor(F.col("event_year") / 10) * 10).cast("int")
)

decade_summary = df_trends.groupBy("decade", "event_type").agg(
    F.count("*").alias("event_count"),
    F.sum("total_damage_amt").alias("total_damage"),
    F.sum("deaths_direct").alias("total_fatalities"),
    F.avg("total_damage_amt").alias("avg_damage_per_event")
).orderBy("decade", F.desc("event_count"))

# Write to Silver
df_classified.write.format("delta").mode("overwrite").option(
    "overwriteSchema", "true"
).save("abfss://lh_silver@onelake.dfs.fabric.microsoft.com/Tables/silver_noaa_storm_classified")

decade_summary.write.format("delta").mode("overwrite").save(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_noaa_storm_decadal_trends"
)

print(f"Storm events classified: {df_classified.count():,}")
print(f"CATASTROPHIC events: {df_classified.filter(F.col('severity_tier') == 'CATASTROPHIC').count():,}")

# Databricks notebook source
# MAGIC %md
# MAGIC # Seasonal Pattern Detection
# MAGIC Rolling window statistics to identify storm season boundary shifts

# COMMAND ----------

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

# Load Silver classified storms
df_storms = spark.read.format("delta").load(
    "abfss://lh_silver@onelake.dfs.fabric.microsoft.com/Tables/silver_noaa_storm_classified"
)

# --- Monthly Event Distribution by Decade ---
df_monthly = df_storms.withColumn(
    "event_year", F.year("begin_date")
).withColumn(
    "event_month", F.month("begin_date")
).withColumn(
    "decade",
    (F.floor(F.col("event_year") / 10) * 10).cast("int")
)

# Focus on key severe weather types
severe_types = ["Tornado", "Hail", "Thunderstorm Wind", "Hurricane", "Flash Flood"]

df_seasonal = df_monthly.filter(
    F.col("event_type").isin(severe_types)
).groupBy("decade", "event_type", "event_month").agg(
    F.count("*").alias("event_count"),
    F.sum("total_damage_amt").alias("total_damage"),
    F.avg("total_damage_amt").alias("avg_damage")
)

# --- Season Boundary Detection ---
# Calculate the month containing the 10th and 90th percentile of events per type/decade
type_decade_window = Window.partitionBy("decade", "event_type").orderBy("event_month")

df_cumulative = df_seasonal.withColumn(
    "cumulative_events",
    F.sum("event_count").over(type_decade_window)
)

total_by_type_decade = df_seasonal.groupBy("decade", "event_type").agg(
    F.sum("event_count").alias("total_events")
)

df_percentile = df_cumulative.join(
    total_by_type_decade, ["decade", "event_type"]
).withColumn(
    "cumulative_pct",
    F.col("cumulative_events") / F.col("total_events")
)

# Season start = first month crossing 10% cumulative, end = first crossing 90%
season_start = df_percentile.filter(
    F.col("cumulative_pct") >= 0.10
).groupBy("decade", "event_type").agg(
    F.min("event_month").alias("season_start_month")
)

season_end = df_percentile.filter(
    F.col("cumulative_pct") >= 0.90
).groupBy("decade", "event_type").agg(
    F.min("event_month").alias("season_end_month")
)

df_season_bounds = season_start.join(
    season_end, ["decade", "event_type"]
).withColumn(
    "season_length_months",
    F.col("season_end_month") - F.col("season_start_month") + 1
).orderBy("event_type", "decade")

# Write Gold
df_season_bounds.write.format("delta").mode("overwrite").save(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_noaa_season_boundaries"
)

print("Season boundary shifts by decade:")
df_season_bounds.show(30, truncate=False)

# Databricks notebook source
# MAGIC %md
# MAGIC # Climate Anomaly Monitoring
# MAGIC Computing departure-from-normal metrics for temperature and precipitation

# COMMAND ----------

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

# Load observations and normals
df_obs = spark.read.format("delta").load(
    "abfss://lh_silver@onelake.dfs.fabric.microsoft.com/Tables/silver_noaa_daily_observations"
)
df_normals = spark.read.format("delta").load(
    "abfss://lh_silver@onelake.dfs.fabric.microsoft.com/Tables/silver_noaa_climate_normals"
)

# --- Temperature Anomaly ---
# Join daily observations with monthly normals by station and month
df_temp = df_obs.withColumn(
    "obs_month", F.month("observation_date")
).withColumn(
    "obs_year", F.year("observation_date")
)

df_anomaly = df_temp.join(
    df_normals,
    on=[
        df_temp.station_id == df_normals.station_id,
        df_temp.obs_month == df_normals.month
    ],
    how="inner"
).select(
    df_temp.station_id,
    df_temp.station_name,
    df_temp.state,
    df_temp.obs_year,
    df_temp.obs_month,
    df_temp.avg_temp_f,
    df_normals.normal_avg_temp_f,
    df_temp.total_precip_in,
    df_normals.normal_precip_in
)

# Compute departures
df_departures = df_anomaly.withColumn(
    "temp_anomaly_f",
    F.round(F.col("avg_temp_f") - F.col("normal_avg_temp_f"), 2)
).withColumn(
    "precip_anomaly_in",
    F.round(F.col("total_precip_in") - F.col("normal_precip_in"), 2)
).withColumn(
    "precip_pct_normal",
    F.round(F.col("total_precip_in") / F.col("normal_precip_in") * 100, 1)
)

# --- Annual State-Level Anomaly Summary ---
state_annual = df_departures.groupBy("state", "obs_year").agg(
    F.avg("temp_anomaly_f").alias("avg_temp_anomaly"),
    F.avg("precip_anomaly_in").alias("avg_precip_anomaly"),
    F.count("*").alias("observation_count"),
    F.sum(F.when(F.col("temp_anomaly_f") > 3.0, 1).otherwise(0)).alias("extreme_warm_months"),
    F.sum(F.when(F.col("temp_anomaly_f") < -3.0, 1).otherwise(0)).alias("extreme_cold_months")
).withColumn(
    "warming_trend",
    F.when(F.col("avg_temp_anomaly") > 2.0, "SIGNIFICANT_WARMING")
    .when(F.col("avg_temp_anomaly") > 0.5, "MODERATE_WARMING")
    .when(F.col("avg_temp_anomaly") < -0.5, "COOLING")
    .otherwise("NEAR_NORMAL")
)

# --- 10-Year Rolling Trend ---
year_window = Window.partitionBy("state").orderBy("obs_year") \
    .rowsBetween(-9, 0)

state_trend = state_annual.withColumn(
    "rolling_10yr_temp_anomaly",
    F.round(F.avg("avg_temp_anomaly").over(year_window), 2)
)

# Write Gold
state_trend.write.format("delta").mode("overwrite").save(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_noaa_climate_anomaly_trends"
)

print(f"State-year anomaly records: {state_trend.count():,}")

# Cross-domain: Climate anomaly impact on crop production
df_climate = spark.read.format("delta").load(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_noaa_climate_anomaly_trends"
)
df_crops = spark.read.format("delta").load(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_usda_crop_production"
)

# Join climate anomalies with crop yields (lagged 1 year for growing season impact)
df_impact = df_climate.join(
    df_crops,
    on=[
        df_climate.state == df_crops.state_alpha,
        df_climate.obs_year == df_crops.year - 1  # Prior year climate affects next harvest
    ],
    how="inner"
).select(
    df_climate.state,
    df_climate.obs_year.alias("climate_year"),
    "avg_temp_anomaly",
    "avg_precip_anomaly",
    "warming_trend",
    "commodity_desc",
    "yield_per_acre",
    "production_value"
)

# Identify climate-stressed crop outcomes
df_stressed = df_impact.filter(
    (F.col("avg_temp_anomaly") > 2.0) | (F.col("avg_precip_anomaly") < -1.5)
).withColumn(
    "climate_stress_type",
    F.when(
        (F.col("avg_temp_anomaly") > 2.0) & (F.col("avg_precip_anomaly") < -1.5),
        "HOT_DRY"
    ).when(F.col("avg_temp_anomaly") > 2.0, "HOT")
    .otherwise("DRY")
)

df_stressed.groupBy("climate_stress_type", "commodity_desc").agg(
    F.avg("yield_per_acre").alias("avg_stressed_yield"),
    F.count("*").alias("sample_size")
).orderBy("commodity_desc", "climate_stress_type").show(30)