Runbook: Mini Medallion ČHMÚ

import urllib.request, os

url = "https://opendata.chmi.cz/air_quality/now/data/airquality_1h_avg_CZ.csv"
raw = urllib.request.urlopen(url).read()
os.makedirs("/lakehouse/default/Files/raw", exist_ok=True)
with open("/lakehouse/default/Files/raw/chmi_snapshot.csv", "wb") as f:
    f.write(raw)
print(f"OK, {len(raw)} bajtů zapsáno do Files/raw/chmi_snapshot.csv")

csv_to_table = [
    ("Files/raw/chmi_snapshot.csv",     "chmi_snapshot_raw"),
    ("Files/lookups/dim_kraj.csv",      "dim_kraj_raw"),
    ("Files/lookups/dim_pollutant.csv", "dim_pollutant_raw"),
    ("Files/lookups/dim_station.csv",   "dim_station_raw"),
]

for path, table in csv_to_table:
    df = (spark.read
          .option("header", "true")
          .option("ignoreLeadingWhiteSpace", "true")  # CHMI má ", " za čárkou
          .option("inferSchema", "true")
          .csv(path))
    df.write.format("delta").mode("overwrite").saveAsTable(table)
    print(f"  {table}: {df.count()} rows")

chmi_snapshot_raw: ~470 rows
dim_kraj_raw:        14 rows
dim_pollutant_raw:   10 rows
dim_station_raw:    201 rows

CREATE SCHEMA bronze;
GO

CREATE TABLE bronze.chmi_measurement (
    idRegistration INT,
    startTime VARCHAR(30),
    idValueType INT,
    value DECIMAL(10,3),
    _source_url VARCHAR(200),
    _ingested_at DATETIME2(6)
);

INSERT INTO bronze.chmi_measurement (idRegistration, startTime, idValueType, value)
SELECT idRegistration, startTime, idValueType, value
FROM DP600.dbo.chmi_snapshot_raw;

UPDATE bronze.chmi_measurement
SET _ingested_at = SYSDATETIME(),
    _source_url = 'https://opendata.chmi.cz/air_quality/now/data/airquality_1h_avg_CZ.csv'
WHERE _ingested_at IS NULL;

SELECT COUNT(*) AS bronze_rows FROM bronze.chmi_measurement;

CREATE SCHEMA silver;
GO

CREATE TABLE silver.dim_kraj (
    kraj_id SMALLINT NOT NULL,
    kraj_kod VARCHAR(3),
    kraj_nazev VARCHAR(50),
    nuts_kod VARCHAR(5)
);

CREATE TABLE silver.dim_pollutant (
    id_value_type SMALLINT NOT NULL,
    kod VARCHAR(10),
    nazev_cs VARCHAR(50),
    nazev_en VARCHAR(50),
    jednotka VARCHAR(10),
    limit_eu_year_avg DECIMAL(6,2),
    limit_eu_24h DECIMAL(6,2),
    note VARCHAR(200)
);

CREATE TABLE silver.dim_station (
    id_registration INT NULL,
    loc_code VARCHAR(4) NOT NULL,
    nazev VARCHAR(80),
    mesto VARCHAR(50),
    kraj_id SMALLINT,
    lat DECIMAL(7,4),
    lon DECIMAL(7,4),
    nadm_vyska_m INT NULL,
    typ_stanice VARCHAR(20),
    loc_owner VARCHAR(20),
    classification VARCHAR(20),
    id_confidence VARCHAR(20)
);

INSERT INTO silver.dim_kraj
SELECT CAST(kraj_id AS SMALLINT), kraj_kod, kraj_nazev, nuts_kod
FROM DP600.dbo.dim_kraj_raw;

INSERT INTO silver.dim_pollutant
SELECT CAST(id_value_type AS SMALLINT), kod, nazev_cs, nazev_en, jednotka,
       TRY_CAST(limit_eu_year_avg AS DECIMAL(6,2)),
       TRY_CAST(limit_eu_24h AS DECIMAL(6,2)),
       note
FROM DP600.dbo.dim_pollutant_raw;

INSERT INTO silver.dim_station
SELECT TRY_CAST(id_registration AS INT), loc_code, nazev, mesto,
       CAST(kraj_id AS SMALLINT),
       TRY_CAST(lat AS DECIMAL(7,4)),
       TRY_CAST(lon AS DECIMAL(7,4)),
       TRY_CAST(nadm_vyska_m AS INT),
       typ_stanice, loc_owner, classification, id_confidence
FROM DP600.dbo.dim_station_raw;

SELECT 'kraj' AS tbl, COUNT(*) AS rows FROM silver.dim_kraj
UNION ALL SELECT 'pollutant', COUNT(*) FROM silver.dim_pollutant
UNION ALL SELECT 'station', COUNT(*) FROM silver.dim_station;

CREATE TABLE silver.fact_measurement (
    id_registration INT,
    loc_code VARCHAR(4) NULL,
    mereni_ts DATETIME2(6),
    datum DATE,
    hodina SMALLINT,
    id_value_type SMALLINT,
    value DECIMAL(10,3),
    kraj_id SMALLINT,
    kraj_nazev VARCHAR(50),
    mesto VARCHAR(50),
    station_nazev VARCHAR(80),
    typ_stanice VARCHAR(20),
    pollutant_kod VARCHAR(10),
    pollutant_nazev VARCHAR(50),
    jednotka VARCHAR(10),
    je_nad_limitem BIT,
    _silver_loaded_at DATETIME2(6)
);

CREATE OR ALTER PROCEDURE silver.sp_load_fact_measurement
AS
BEGIN
    TRUNCATE TABLE silver.fact_measurement;

    INSERT INTO silver.fact_measurement
    (id_registration, loc_code, mereni_ts, datum, hodina,
     id_value_type, value,
     kraj_id, kraj_nazev, mesto, station_nazev, typ_stanice,
     pollutant_kod, pollutant_nazev, jednotka, je_nad_limitem,
     _silver_loaded_at)
    SELECT
        b.idRegistration,
        s.loc_code,
        TRY_CONVERT(DATETIME2(6), REPLACE(b.startTime, 'Z', '')),
        TRY_CONVERT(DATE, REPLACE(b.startTime, 'Z', '')),
        DATEPART(HOUR, TRY_CONVERT(DATETIME2(6), REPLACE(b.startTime, 'Z', ''))),
        b.idValueType,
        b.value,
        ISNULL(s.kraj_id, -1),
        ISNULL(k.kraj_nazev, 'Unmapped'),
        ISNULL(s.mesto, 'Unknown'),
        ISNULL(s.nazev, 'ID ' + CAST(b.idRegistration AS VARCHAR(20))),
        ISNULL(s.typ_stanice, 'unknown'),
        p.kod,
        p.nazev_cs,
        p.jednotka,
        CASE
            WHEN p.limit_eu_year_avg IS NOT NULL AND b.value > p.limit_eu_year_avg THEN 1
            WHEN p.limit_eu_24h IS NOT NULL AND b.value > p.limit_eu_24h THEN 1
            ELSE 0
        END,
        SYSDATETIME()
    FROM bronze.chmi_measurement b
    LEFT JOIN silver.dim_station s ON s.id_registration = b.idRegistration
    LEFT JOIN silver.dim_kraj k ON k.kraj_id = s.kraj_id
    LEFT JOIN silver.dim_pollutant p ON p.id_value_type = b.idValueType
    WHERE b.value IS NOT NULL;

    SELECT COUNT(*) AS silver_rows,
           SUM(CASE WHEN id_registration IS NULL OR kraj_id = -1 THEN 1 ELSE 0 END) AS unmapped
    FROM silver.fact_measurement;
END;
GO

EXEC silver.sp_load_fact_measurement;

CREATE SCHEMA gold;
GO

CREATE TABLE gold.fact_kvalita_ovzdusi (
    datum DATE,
    hodina SMALLINT,
    kraj_id SMALLINT,
    kraj_nazev VARCHAR(50),
    pollutant_kod VARCHAR(10),
    pollutant_nazev VARCHAR(50),
    jednotka VARCHAR(10),
    pocet_stanic INT,
    prum_value DECIMAL(10,3),
    max_value DECIMAL(10,3),
    min_value DECIMAL(10,3),
    pocet_nad_limitem INT,
    _gold_loaded_at DATETIME2(6)
);

CREATE OR ALTER PROCEDURE gold.sp_load_kvalita_ovzdusi
AS
BEGIN
    TRUNCATE TABLE gold.fact_kvalita_ovzdusi;

    INSERT INTO gold.fact_kvalita_ovzdusi
    SELECT
        datum, hodina,
        kraj_id, kraj_nazev,
        pollutant_kod, pollutant_nazev, jednotka,
        COUNT(DISTINCT id_registration),
        AVG(value), MAX(value), MIN(value),
        SUM(CAST(je_nad_limitem AS INT)),
        SYSDATETIME()
    FROM silver.fact_measurement
    WHERE kraj_id <> -1
      AND pollutant_kod IS NOT NULL
    GROUP BY datum, hodina, kraj_id, kraj_nazev,
             pollutant_kod, pollutant_nazev, jednotka;

    SELECT COUNT(*) AS gold_rows FROM gold.fact_kvalita_ovzdusi;
END;
GO

EXEC gold.sp_load_kvalita_ovzdusi;

SELECT TOP 50 * FROM gold.fact_kvalita_ovzdusi;