Runbook: Big Medallion OpenAQ (2 roky)

import urllib.request, json, os, gzip, io
import pandas as pd
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

API_KEY = "REPLACE_WITH_YOUR_OPENAQ_KEY"

def fetch_api(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_api("/locations", "countries_id=49&limit=1000")
cz_loc_ids = [L["id"] for L in locations["results"]]
print(f"CZ locations: {len(cz_loc_ids)}")

loc_meta = pd.DataFrame([{
    "location_id": L["id"],
    "name": L.get("name"),
    "city": L.get("locality") or "",
    "lat": (L.get("coordinates") or {}).get("latitude"),
    "lon": (L.get("coordinates") or {}).get("longitude"),
    "owner_name": L.get("owner", {}).get("name", ""),
    "is_mobile": L.get("isMobile", False),
} for L in locations["results"]])

os.makedirs("/lakehouse/default/Files/historical", exist_ok=True)
loc_meta.to_parquet("/lakehouse/default/Files/historical/cz_locations.parquet", index=False)
print(f"Locations metadata: {len(loc_meta)} rows uloženo")

import subprocess
subprocess.run(['pip', 'install', 'boto3', '--quiet'], check=True)

import boto3
from botocore import UNSIGNED
from botocore.client import Config
from concurrent.futures import ThreadPoolExecutor, as_completed
import io, gzip, time
import pandas as pd

s3 = boto3.client('s3', config=Config(signature_version=UNSIGNED))
YEARS = (2023, 2024)

def list_keys_for_loc(loc_id):
    keys = []
    paginator = s3.get_paginator('list_objects_v2')
    for year in YEARS:
        prefix = f'records/csv.gz/locationid={loc_id}/year={year}/'
        for page in paginator.paginate(Bucket='openaq-data-archive', Prefix=prefix):
            for obj in page.get('Contents', []):
                keys.append(obj['Key'])
    return keys

print("=== A: enumerace S3 keys ===")
start = time.time()
all_keys = []
with ThreadPoolExecutor(max_workers=20) as ex:
    futures = {ex.submit(list_keys_for_loc, loc): loc for loc in cz_loc_ids}
    for i, f in enumerate(as_completed(futures)):
        all_keys.extend(f.result())
        if (i+1) % 30 == 0:
            print(f"  enumerated {i+1}/{len(cz_loc_ids)} stanic, keys: {len(all_keys)}")

print(f"\nTotal keys k downloadu: {len(all_keys)} ({time.time()-start:.0f}s)")

def download_key(key):
    try:
        obj = s3.get_object(Bucket='openaq-data-archive', Key=key)
        with gzip.GzipFile(fileobj=io.BytesIO(obj['Body'].read())) as f:
            df = pd.read_csv(f)
            df['_location_id'] = int(key.split('locationid=')[1].split('/')[0])
            return df
    except Exception:
        return None

print("\n=== B: download ===")
start = time.time()
all_dfs = []
with ThreadPoolExecutor(max_workers=30) as ex:
    futures = [ex.submit(download_key, k) for k in all_keys]
    for i, f in enumerate(as_completed(futures)):
        df = f.result()
        if df is not None and not df.empty:
            all_dfs.append(df)
        if (i+1) % 1000 == 0:
            print(f"  progress: {i+1}/{len(all_keys)} ({time.time()-start:.0f}s, {len(all_dfs)} success)")

print(f"\nDownloaded {len(all_dfs)} files v {time.time()-start:.0f}s")
print(f"Total rows: {sum(len(d) for d in all_dfs):,}")

YEARS = (2024,)

cz_loc_ids = cz_loc_ids[:30]

df_all = pd.concat(all_dfs, ignore_index=True)
print(f"Concatenated: {df_all.shape}")
print(df_all.dtypes)
print(df_all.head(3))

df_all.columns = [c.strip() for c in df_all.columns]

if "datetime" in df_all.columns:
    df_all["datetime"] = pd.to_datetime(df_all["datetime"], errors="coerce")

out_path = "/lakehouse/default/Files/historical/openaq_cz_measurements_2y.parquet"
df_all.to_parquet(out_path, index=False, compression="snappy")
print(f"Saved {len(df_all):,} rows to {out_path}")
print(f"File size: {os.path.getsize(out_path)/1024/1024:.1f} MB")

del all_dfs, df_all

import polars as pl
combined = pl.concat([pl.from_pandas(d) for d in all_dfs])
combined.write_parquet(out_path)

df_meas = spark.read.parquet("Files/historical/openaq_cz_measurements_2y.parquet")
df_meas.write.format("delta").mode("overwrite").saveAsTable("openaq_meas_full_raw")
print(f"openaq_meas_full_raw: {df_meas.count():,} rows")

df_loc = spark.read.parquet("Files/historical/cz_locations.parquet")
df_loc.write.format("delta").mode("overwrite").saveAsTable("openaq_loc_full_raw")
print(f"openaq_loc_full_raw: {df_loc.count()} rows")

-- Schema guard (idempotentní)
IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = 'bronze')
    EXEC('CREATE SCHEMA bronze');
GO

-- Tables (drop + create = ekvivalent CREATE OR REPLACE)
DROP TABLE IF EXISTS bronze.openaq_meas_full;
CREATE TABLE bronze.openaq_meas_full (
    location_id INT,
    sensors_id INT,
    parameter_name VARCHAR(50),
    parameter_units VARCHAR(20),
    value DECIMAL(18,6),
    datetime_from DATETIME2(6),
    datetime_to DATETIME2(6),
    lat DECIMAL(10,6),
    lon DECIMAL(10,6),
    _ingested_at DATETIME2(6)
);

DROP TABLE IF EXISTS bronze.openaq_loc_full;
CREATE TABLE bronze.openaq_loc_full (
    location_id INT,
    name VARCHAR(200),
    city VARCHAR(100),
    lat DECIMAL(10,6),
    lon DECIMAL(10,6),
    owner_name VARCHAR(200),
    is_mobile BIT,
    _ingested_at DATETIME2(6)
);
GO

INSERT INTO bronze.openaq_meas_full (location_id, parameter_name, parameter_units, value, datetime_from, lat, lon, _ingested_at)
SELECT
    _location_id,
    [parameter],
    [units],
    [value],
    TRY_CONVERT(DATETIME2(6), [datetime]),
    [lat], [lon],
    SYSDATETIME()
FROM LH_OpenAQ_Full.dbo.openaq_meas_full_raw;

INSERT INTO bronze.openaq_loc_full (location_id, name, city, lat, lon, owner_name, is_mobile, _ingested_at)
SELECT location_id, name, city, lat, lon, owner_name, is_mobile, SYSDATETIME()
FROM LH_OpenAQ_Full.dbo.openaq_loc_full_raw;

SELECT 'meas' AS tbl, COUNT(*) AS rows FROM bronze.openaq_meas_full
UNION ALL SELECT 'loc', COUNT(*) FROM bronze.openaq_loc_full;

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)
);

CREATE TABLE silver.fact_measurement (
    location_id INT,
    city VARCHAR(100),
    parameter_name VARCHAR(50),
    parameter_units VARCHAR(20),
    value DECIMAL(18,6),
    mereni_ts DATETIME2(6),
    datum DATE,
    rok SMALLINT,
    mesic SMALLINT,
    den_v_tydnu SMALLINT,
    hodina SMALLINT,
    _silver_loaded_at DATETIME2(6)
);
GO

INSERT INTO silver.dim_location
SELECT DISTINCT location_id, name, city, lat, lon FROM bronze.openaq_loc_full;

INSERT INTO silver.fact_measurement
SELECT
    m.location_id,
    l.city,
    m.parameter_name,
    m.parameter_units,
    m.value,
    m.datetime_from,
    CAST(m.datetime_from AS DATE),
    YEAR(m.datetime_from),
    MONTH(m.datetime_from),
    DATEPART(WEEKDAY, m.datetime_from),
    DATEPART(HOUR, m.datetime_from),
    SYSDATETIME()
FROM bronze.openaq_meas_full m
LEFT JOIN bronze.openaq_loc_full l ON l.location_id = m.location_id
WHERE m.value IS NOT NULL
  AND m.value > 0  -- filter zjevně chybné
  AND m.datetime_from >= '2023-01-01';

CREATE SCHEMA gold;
GO

-- Gold 1: měsíční trendy per city × parameter
CREATE TABLE gold.fact_monthly_trends (
    rok SMALLINT,
    mesic SMALLINT,
    city VARCHAR(100),
    parameter_name VARCHAR(50),
    pocet_meri INT,
    avg_value DECIMAL(18,6),
    max_value DECIMAL(18,6)
);

-- Gold 2: hodina × den (heatmap pollution patterns)
CREATE TABLE gold.fact_hourly_pattern (
    parameter_name VARCHAR(50),
    den_v_tydnu SMALLINT,
    hodina SMALLINT,
    avg_value DECIMAL(18,6),
    pocet_meri INT
);

-- Gold 3: top stanice (ranking)
CREATE TABLE gold.fact_station_summary (
    location_id INT,
    name VARCHAR(200),
    city VARCHAR(100),
    parameter_name VARCHAR(50),
    pocet_meri INT,
    avg_value DECIMAL(18,6),
    p95_value DECIMAL(18,6)
);
GO

INSERT INTO gold.fact_monthly_trends
SELECT rok, mesic, city, parameter_name,
       COUNT(*), AVG(value), MAX(value)
FROM silver.fact_measurement
WHERE city IS NOT NULL AND city <> ''
GROUP BY rok, mesic, city, parameter_name;

INSERT INTO gold.fact_hourly_pattern
SELECT parameter_name, den_v_tydnu, hodina,
       AVG(value), COUNT(*)
FROM silver.fact_measurement
GROUP BY parameter_name, den_v_tydnu, hodina;

INSERT INTO gold.fact_station_summary
SELECT
    f.location_id,
    l.name,
    f.city,
    f.parameter_name,
    COUNT(*),
    AVG(f.value),
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY f.value) OVER (PARTITION BY f.location_id, f.parameter_name)
FROM silver.fact_measurement f
LEFT JOIN silver.dim_location l ON l.location_id = f.location_id
GROUP BY f.location_id, l.name, f.city, f.parameter_name;

SELECT TOP 20 * FROM gold.fact_station_summary
WHERE parameter_name = 'pm10'
ORDER BY avg_value DESC;