Runbook: Mini Medallion OpenAQ

import urllib.request, json, os
from pyspark.sql import functions as F

API_KEY = "REPLACE_WITH_YOUR_OPENAQ_KEY"  # z api.openaq.org/register

def fetch_openaq(path, params=""):
    url = f"https://api.openaq.org/v3{path}?{params}"
    req = urllib.request.Request(url, headers={"X-API-Key": API_KEY})
    return json.loads(urllib.request.urlopen(req).read())

locations = fetch_openaq("/locations", "countries_id=49&limit=1000")
print(f"Loaded {locations['meta']['found']} CZ locations")

os.makedirs("/lakehouse/default/Files/raw", exist_ok=True)
with open("/lakehouse/default/Files/raw/openaq_locations.json", "w") as f:
    json.dump(locations, f, ensure_ascii=False)
print("Locations saved to Files/raw/openaq_locations.json")

all_measurements = []
for loc in locations["results"][:50]:  # limit pro demo
    loc_id = loc["id"]
    try:
        m = fetch_openaq(f"/locations/{loc_id}/latest")
        for r in m.get("results", []):
            r["location_id"] = loc_id
            r["location_name"] = loc.get("name", "")
            all_measurements.append(r)
    except Exception as e:
        print(f"  skip loc {loc_id}: {e}")

print(f"Loaded {len(all_measurements)} measurements")

with open("/lakehouse/default/Files/raw/openaq_measurements.json", "w") as f:
    json.dump(all_measurements, f, ensure_ascii=False)

loc_rows = []
for L in locations["results"]:
    coords = L.get("coordinates") or {}
    loc_rows.append({
        "location_id": L["id"],
        "name": L.get("name"),
        "city": (L.get("locality") or ""),
        "country": L.get("country", {}).get("code", ""),
        "lat": coords.get("latitude"),
        "lon": coords.get("longitude"),
        "owner_name": L.get("owner", {}).get("name", ""),
        "is_mobile": L.get("isMobile", False),
        "sensors_count": len(L.get("sensors", []) or []),
    })

import pandas as pd
df_loc = pd.DataFrame(loc_rows)
sdf_loc = spark.createDataFrame(df_loc)
sdf_loc.write.format("delta").mode("overwrite").saveAsTable("openaq_locations_raw")
print(f"openaq_locations_raw: {sdf_loc.count()} rows")

meas_rows = []
for m in all_measurements:
    period = m.get("period") or {}
    coords = m.get("coordinates") or {}
    meas_rows.append({
        "location_id": m.get("location_id"),
        "location_name": m.get("location_name"),
        "parameter_name": (m.get("parameter") or {}).get("name", ""),
        "parameter_units": (m.get("parameter") or {}).get("units", ""),
        "value": m.get("value"),
        "datetime_utc": (period.get("datetimeFrom") or {}).get("utc", ""),
        "lat": coords.get("latitude"),
        "lon": coords.get("longitude"),
    })

df_m = pd.DataFrame(meas_rows)
sdf_m = spark.createDataFrame(df_m)
sdf_m.write.format("delta").mode("overwrite").saveAsTable("openaq_measurements_raw")
print(f"openaq_measurements_raw: {sdf_m.count()} rows")

CREATE SCHEMA bronze;
GO

CREATE TABLE bronze.openaq_locations (
    location_id INT,
    name VARCHAR(200),
    city VARCHAR(100),
    country VARCHAR(5),
    lat DECIMAL(10,6),
    lon DECIMAL(10,6),
    owner_name VARCHAR(200),
    is_mobile BIT,
    sensors_count INT,
    _ingested_at DATETIME2(6)
);

CREATE TABLE bronze.openaq_measurements (
    location_id INT,
    location_name VARCHAR(200),
    parameter_name VARCHAR(50),
    parameter_units VARCHAR(20),
    value DECIMAL(18,6),
    datetime_utc VARCHAR(40),
    lat DECIMAL(10,6),
    lon DECIMAL(10,6),
    _ingested_at DATETIME2(6)
);
GO

INSERT INTO bronze.openaq_locations
(location_id, name, city, country, lat, lon, owner_name, is_mobile, sensors_count, _ingested_at)
SELECT location_id, name, city, country, lat, lon, owner_name, is_mobile, sensors_count, SYSDATETIME()
FROM LH_OpenAQ.dbo.openaq_locations_raw;

INSERT INTO bronze.openaq_measurements
(location_id, location_name, parameter_name, parameter_units, value, datetime_utc, lat, lon, _ingested_at)
SELECT location_id, location_name, parameter_name, parameter_units, value, datetime_utc, lat, lon, SYSDATETIME()
FROM LH_OpenAQ.dbo.openaq_measurements_raw;

SELECT 'locations' AS tbl, COUNT(*) AS rows FROM bronze.openaq_locations
UNION ALL SELECT 'measurements', COUNT(*) FROM bronze.openaq_measurements;

CREATE SCHEMA silver;
GO

CREATE TABLE silver.dim_location (
    location_id INT NOT NULL,
    name VARCHAR(200),
    city VARCHAR(100),
    lat DECIMAL(10,6),
    lon DECIMAL(10,6),
    is_mobile BIT,
    owner_name VARCHAR(200)
);

CREATE TABLE silver.dim_parameter (
    parameter_name VARCHAR(50) NOT NULL,
    parameter_units VARCHAR(20),
    description VARCHAR(200)
);

CREATE TABLE silver.fact_measurement (
    location_id INT,
    location_name VARCHAR(200),
    city VARCHAR(100),
    parameter_name VARCHAR(50),
    parameter_units VARCHAR(20),
    value DECIMAL(18,6),
    mereni_ts DATETIME2(6),
    datum DATE,
    hodina SMALLINT,
    lat DECIMAL(10,6),
    lon DECIMAL(10,6),
    _silver_loaded_at DATETIME2(6)
);
GO

-- dim_location
INSERT INTO silver.dim_location
SELECT DISTINCT location_id, name, city, lat, lon, is_mobile, owner_name
FROM bronze.openaq_locations;

-- dim_parameter (auto-extract z měření)
INSERT INTO silver.dim_parameter
SELECT DISTINCT parameter_name, parameter_units, parameter_name
FROM bronze.openaq_measurements
WHERE parameter_name IS NOT NULL;

-- fact_measurement
INSERT INTO silver.fact_measurement
SELECT
    m.location_id,
    m.location_name,
    l.city,
    m.parameter_name,
    m.parameter_units,
    m.value,
    TRY_CONVERT(DATETIME2(6), REPLACE(m.datetime_utc, 'Z', '')),
    TRY_CONVERT(DATE, REPLACE(m.datetime_utc, 'Z', '')),
    DATEPART(HOUR, TRY_CONVERT(DATETIME2(6), REPLACE(m.datetime_utc, 'Z', ''))),
    m.lat, m.lon,
    SYSDATETIME()
FROM bronze.openaq_measurements m
LEFT JOIN bronze.openaq_locations l ON l.location_id = m.location_id
WHERE m.value IS NOT NULL;

SELECT 'dim_location' AS tbl, COUNT(*) AS rows FROM silver.dim_location
UNION ALL SELECT 'dim_parameter', COUNT(*) FROM silver.dim_parameter
UNION ALL SELECT 'fact_measurement', COUNT(*) FROM silver.fact_measurement;

CREATE SCHEMA gold;
GO

CREATE TABLE gold.fact_air_quality (
    datum DATE,
    hodina SMALLINT,
    city VARCHAR(100),
    parameter_name VARCHAR(50),
    parameter_units VARCHAR(20),
    pocet_stanic INT,
    avg_value DECIMAL(18,6),
    max_value DECIMAL(18,6),
    min_value DECIMAL(18,6)
);
GO

INSERT INTO gold.fact_air_quality
SELECT
    datum, hodina, city, parameter_name, parameter_units,
    COUNT(DISTINCT location_id),
    AVG(value), MAX(value), MIN(value)
FROM silver.fact_measurement
WHERE city IS NOT NULL AND city <> ''
GROUP BY datum, hodina, city, parameter_name, parameter_units;

SELECT TOP 50 * FROM gold.fact_air_quality
ORDER BY avg_value DESC;