Demo: Tři cesty a Mini Medallion

Lektorský scénář pro hlavní demo Day 1+2. Postaveno nad CHMI air quality daty (kvalita ovzduší v ČR, aktuální i historická).

Story arc dema:

1. Story setup — co je Lakehouse, end-to-end flow, ukázka cílového PBI reportu
2. Bronze, 3 cesty — stejný CHMI snapshot načten třemi způsoby (Notebook / DF Gen2 / Pipeline) do stejné Lakehouse Files zóny
3. Mini Medallion přes Warehouse + SQL procedury — Bronze → Silver → Gold přes T-SQL
4. Closure — Power BI report s mapou krajů + trendy polutantů

Demo se v Day 1 rozprostírá mezi lekce (ne najednou):

• Story setup → před lekcí 1.1 End-to-end overview
• Cesta 1 (Notebook) → během lekce 1.5 Spark
• Cesta 2 (DF Gen2) → během lekce 1.3 Dataflow Gen2
• Cesta 3 (Pipeline) → během lekce 1.4 Pipelines
• Mini Medallion → běží na pozadí, fragmenty napříč Day 2
• Closure PBI → konec Day 1, jako "to je kam míříme v Day 3"

Rozdělení:

• Demo tenant = lektorův, F2 kapacita, start/stop mode (pauza mimo demo bloky)
• Účastníci = vlastní tenant, lektor do něj nepřistupuje
• Lektor demuje, účastníci sledují — žádné share screen na účastníkův tenant
• F64 trial workspace v lektorově tenantu jako bonus srovnání rychlosti, pokud je čas

Důsledky pro demo:

1. Před každým demo blokem resume kapacitu F2 (10–20 s lag než je ready)
2. Po dni demo pause kapacitu → CU billing stop
3. Anti-pattern: pokud potřebuješ ukázat publish flow, použij screenshot, ne live

A. Lookup tabulky

Tři lookup CSV (dim_kraj, dim_pollutant, dim_station) jsou hotové ke stažení v runbooku Mini Medallion ČHMÚ. dim_station.csv je odvozený z oficiálního registru stanic ČHMÚ (GeoPackage na geoportal.gov.cz, CC BY 4.0) — ~200 stanic s GPS, krajem a typem.

Hotový dim_station.csv jde přepoužít pro každý další kurz — registr se mění zřídka.

B. Workspace setup

Tenant:    <tenant lektora>
Workspace: DP600_Demo_CHMI (F2 kapacita, start/stop)
Lakehouse: ovzdusi_demo
Warehouse: ovzdusi_dw
Optional:  DP600_Demo_F64_Compare (na F64 trial pro speed-up srovnání)

C. Files struktura v Lakehouse

Files/
├── lookups/
│   ├── dim_kraj.csv
│   ├── dim_pollutant.csv
│   └── dim_station.csv (po validaci)
├── raw/
│   └── (Bronze ingest cíl — plní se v demu)
└── recent/
    └── (volitelně pre-stáhnutý 30denní snapshot pro fallback)

Upload lookup CSVs jednou před kurzem přes Lakehouse Explorer → Upload.

D. Pre-buildy pro fallback

E. Pre-run dopředu pro Capacity Metrics

Den před kurzem (24 h zpět = data v Metrics app už budou):

1. Spustit všechny 3 Bronze cesty + Silver + Gold v F2 workspace
2. Ráno před kurzem screenshot z Capacity Metrics

Talk track:

"Co budeme dělat. Dataset = aktuální kvalita ovzduší v ČR od ČHMÚ. Hodinové průměry pro všechny stanice (~470 stanic), ~10 polutantů: PM10, NO2, ozon, SO2, benzen, CO. Cílem je dostat se z živého CSV URL k Power BI dashboardu, který odpoví na otázky typu 'Který kraj má nejhorší ovzduší dnes' nebo 'Kdy je v Brně nejvíc smogu'.

Cestou ukážu tři různé způsoby, jak nahrát stejná data: notebook, Dataflow, Pipeline. Pokaždé se dostaneme na stejné místo. Pak postavíme mini medallion — Bronze (raw), Silver (čištěno), Gold (agregováno) — přes SQL procedury do Warehouse. Na závěr Day 1 ukážu výsledný report.

Klíčový výhoda CHMI dat: jeden URL, žádná autentizace, žádná extrakce, čistý CSV. Tenhle dataset si po kurzu můžete vzít a přepoužít na vlastní firemní data — workflow je úplně stejný.

Demo poběží na F2 kapacitě — to je nejmenší placená kapacita Fabricu, asi 250 USD/měsíc. Uvidíte, jak se to chová. Spark se na F2 startuje 1–2 minuty — během toho budu povídat o architektuře."

URL: https://opendata.chmi.cz/air_quality/now/data/airquality_1h_avg_CZ.csv

Charakter: snapshot poslední hodinové průměry, ~470 řádků (1 řádek = 1 stanice × 1 polutant), refresh každou hodinu.

Schema (4 sloupce, comma-delimited, UTF-8, leading whitespace):

df = pd.read_csv(url, skipinitialspace=True)  # ne sep=";", ne encoding=

Pro multi-hour history použij recent/ adresář s rolling oknem:

https://opendata.chmi.cz/air_quality/recent/data/airquality_1h_avg_CZ_YYYYMMDDHH.csv

(Naming convention nutno ověřit při validation runu.)

Talk track úvod:

"První cesta — programmatic. Notebook, PySpark. Tohle volíte když máte data engineera, vyhovuje vám kód, potřebujete logiku co Dataflow neumí, nebo chcete reusable knihovnu transformací."

Notebook: Demo_01_Bronze_PySpark

from pyspark.sql.functions import col, lit, current_timestamp

import urllib.request
url = "https://opendata.chmi.cz/air_quality/now/data/airquality_1h_avg_CZ.csv"
raw = urllib.request.urlopen(url).read().decode("utf-8")
print(f"Staženo {len(raw)} bajtů")

mssparkutils.fs.put(
    "Files/raw/chmi_snapshot.csv",
    raw,
    overwrite=True,
)

df = (spark.read
    .option("header", "true")
    .option("delimiter", ",")
    .option("inferSchema", "true")
    .option("ignoreLeadingWhiteSpace", "true")  # ekvivalent skipinitialspace
    .csv("Files/raw/chmi_snapshot.csv"))

print(f"Načteno řádků: {df.count():,}")
df.printSchema()
df.show(5)

df_bronze = (df
    .withColumn("_source_url", lit(url))
    .withColumn("_ingested_at", current_timestamp())
    .withColumn("_ingest_method", lit("notebook_pyspark"))
)

(df_bronze.write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .saveAsTable("ovzdusi_bronze"))

spark.sql("SELECT COUNT(*) FROM ovzdusi_bronze").show()
spark.sql("DESCRIBE DETAIL ovzdusi_bronze").show(truncate=False)

Co říct během cells:

• Cell 2: "Spark sám HTTPS nečte. Použijeme Python na stažení, mssparkutils na zápis. Trochu šachy, ale produkční scénář."
• Cell 3: "ignoreLeadingWhiteSpace — CHMI má za každou čárkou mezeru. Bez toho by sloupce měly bizarní názvy."
• Cell 4: "_source_url, _ingested_at = lineage. Bronze nikdy bez lineage."

Pure Python alternativa pro F2 šetření:

import pandas as pd, urllib.request
df = pd.read_csv(url, skipinitialspace=True)
from deltalake import write_deltalake
write_deltalake("/lakehouse/default/Tables/ovzdusi_bronze", df, mode="overwrite")

Talk: "Pro 470 řádků je Spark zbytečný overhead. Pure Python notebook startuje instantně. Pro celý DP-600 worth knowing — ne všechno musí být PySpark."

Risky body:

• Spark cold start (1–2 min na F2) — naplánuj povídání o Spark architecture
• Pokud Spark zhavaruje, otevři pre-built notebook a ukaž že "tady už to běželo"

Kdy v Day 1: během lekce 1.3 Dataflow Gen2.

Talk track úvod:

"Druhá cesta — low-code. Dataflow Gen2 = Power Query v Fabricu. Tohle volíte když máte analytika z Power BI, který už zná Power Query, a chcete mu dát ETL nástroj bez kódu. UI je shodné s Power Query Online z Power BI Service."

Click-through:

Klíčové talk pointy:

• "Pod kapotou je M language. Můžete přepnout do Advanced editor a vidět kód."
• "Destination = Lakehouse, ale taky Warehouse, KQL DB nebo SQL DB. Big advantage proti Dataflow Gen1."
• "Fast Copy mode (od 2024) — DF Gen2 detekuje, kdy může jet jako Pipeline copy."
• "Web zdroj je největší výhoda Dataflow Gen2 pro tenhle case. Spark by tu samou věc načítal přes mezikrok do Files. DF Gen2 streamuje rovnou."

Risky body:

• Destination configurace má někdy bug — když "Save & Run" hodí error, vyresetovat: smazat destinaci, znovu nastavit
• Refresh history v jiném submenu — Workspace → tři tečky vedle DF → "View refresh history"

Kdy v Day 1: během lekce 1.4 Pipelines.

Talk track úvod:

"Třetí cesta — orchestrace. Pipeline. Tohle volíte když máte složitější flow: stáhnout z S3, přesunout, vyčistit, zavolat notebook, poslat email. Pipeline je orchestrátor."

Click-through:

Klíčové talk pointy:

• "Pipeline má 30+ aktivit: Copy Data, Notebook, Stored Procedure, ForEach, If, Wait, Web. Když potřebujete větvit, podmínkovat, paralelizovat — pipeline."
• "Schedule trigger v záložce Schedule. Daily ETL by tady stál."
• "Pro CHMI use case: Pipeline by stahovala hodinový snapshot každou hodinu. Schedule Trigger → Recurrence → 1 hour."

Risky body:

• HTTP source někdy potřebuje wait+retry (CHMI občas timeoutuje)
• Schedule UI bývá pomalá k loadnutí

Kdy v Day 2: napříč lekcemi 2.1 (Warehouse start), 2.2 (Load data), 2.3 (Query).

Talk track úvod:

"Teď ukážu, jak to vypadá v praxi pro tým s SQL backgroundem. Místo Spark notebooku napíšeme T-SQL. Bronze → Silver → Gold přes stored procedures ve Warehouse. Pro tým se SQL Server backgroundem je tohle nejnižší overhead — víte všechno, jen syntax v Fabric Warehouse má pár nuancí."

Step 1: Vytvořit Warehouse + Bronze tabulku

-- Bronze landing: mirror CSV struktury
CREATE SCHEMA bronze;
GO

CREATE TABLE bronze.chmi_measurement (
    idRegistration INT,
    startTime VARCHAR(30),       -- ISO 8601 string, parsujeme později
    idValueType INT,
    value DECIMAL(10,3),
    _source_url VARCHAR(200),
    _ingested_at DATETIME2
);

Step 2: Load do Bronze přes COPY INTO

COPY INTO bronze.chmi_measurement (idRegistration, startTime, idValueType, value)
FROM 'https://onelake.dfs.fabric.microsoft.com/<workspace_id>/<lakehouse_id>/Files/raw/chmi_snapshot.csv'
WITH (
    FILE_TYPE = 'CSV',
    FIELDTERMINATOR = ',',
    FIRSTROW = 2,
    ENCODING = 'UTF8'
);

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(*) FROM bronze.chmi_measurement;  -- ~470

Step 3: Lookup tabulky (Silver dependencies)

Lookup CSVs už máš v Files/lookups/ (z pre-kurz přípravy). Načti je:

CREATE SCHEMA silver;
GO

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

COPY INTO silver.dim_kraj
FROM 'https://onelake.dfs.fabric.microsoft.com/<ws>/<lh>/Files/lookups/dim_kraj.csv'
WITH (FILE_TYPE = 'CSV', FIRSTROW = 2, ENCODING = 'UTF8');

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

COPY INTO silver.dim_pollutant
FROM 'https://onelake.dfs.fabric.microsoft.com/<ws>/<lh>/Files/lookups/dim_pollutant.csv'
WITH (FILE_TYPE = 'CSV', FIRSTROW = 2, ENCODING = 'UTF8');

-- dim_station
-- POZOR: ČHMÚ má dva ID systémy.
-- Live API: idRegistration (numeric, např. 10221, 40237)
-- Registry: loc_code (alphabetic 4-letter, např. ABRE, LLIL)
-- Bridge tabulka v open data není. id_registration je NULL pro většinu řádků.

CREATE TABLE silver.dim_station (
    id_registration INT NULL,        -- bridge na live API (většinou NULL)
    loc_code VARCHAR(4) NOT NULL,     -- EOI/EEA kód (de facto PK)
    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),            -- ČHMÚ, ZÚ Ústí nL, SMBrno, atd.
    classification VARCHAR(20),        -- B/U/RN, T/U/C, atd. (typ/zóna/zdroj)
    id_confidence VARCHAR(20)         -- HIGH_BRIDGED, HIGH_REGISTRY
);

COPY INTO silver.dim_station
FROM 'https://onelake.dfs.fabric.microsoft.com/<ws>/<lh>/Files/lookups/dim_station.csv'
WITH (FILE_TYPE = 'CSV', FIRSTROW = 2, ENCODING = 'UTF8');

SELECT 'kraj' AS dim, 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
UNION ALL SELECT 'station_with_bridge', COUNT(*) FROM silver.dim_station WHERE id_registration IS NOT NULL;

Step 4: Stored procedure Bronze → Silver

CREATE OR ALTER PROCEDURE silver.sp_load_fact_measurement
AS
BEGIN
    IF OBJECT_ID('silver.fact_measurement', 'U') IS NOT NULL
        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, REPLACE(b.startTime, 'Z', '')) AS mereni_ts,
        TRY_CONVERT(DATE, REPLACE(b.startTime, 'Z', ''))      AS datum,
        DATEPART(HOUR, TRY_CONVERT(DATETIME2, REPLACE(b.startTime, 'Z', ''))) AS hodina,
        b.idValueType,
        b.value,
        ISNULL(s.kraj_id, -1)       AS kraj_id,
        ISNULL(k.kraj_nazev, N'Unmapped') AS kraj_nazev,
        ISNULL(s.mesto, N'Unknown') AS mesto,
        ISNULL(s.nazev, N'ID ' + CAST(b.idRegistration AS VARCHAR(20))) AS station_nazev,
        ISNULL(s.typ_stanice, 'unknown') AS typ_stanice,
        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 AS je_nad_limitem,
        SYSDATETIME()
    FROM bronze.chmi_measurement b
    LEFT JOIN silver.dim_station s ON s.id_registration = b.idRegistration  -- bridge join
    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_stations,
           SUM(je_nad_limitem) AS over_limit_count
    FROM silver.fact_measurement;
END;
GO

CREATE TABLE silver.fact_measurement (
    id_registration INT,
    loc_code VARCHAR(4) NULL,
    mereni_ts DATETIME2,
    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
);

EXEC silver.sp_load_fact_measurement;

Talk track:

• "TRY_CONVERT místo CONVERT — invalid timestamp neshodí celý insert."
• "LEFT JOIN s ISNULL fallback — kraj_id = -1 a station_nazev = 'ID 12345' pro unmapped stanice. To je realistický pattern: live data má víc stanic než registr, část měření je vždy 'neznámá'. Pro Gold layer pak filtrujeme WHERE kraj_id <> -1 nebo to ukážeme oddělenou kategorií."
• "je_nad_limitem = computed flag. Star schema-friendly, BI tým nemusí psát CASE WHEN v PBI."
• "Druhý SELECT na konci procedury hlásí počet unmapped — sledovat jako data quality metric."

Step 5: Gold — agregace pro PBI

CREATE SCHEMA gold;
GO

CREATE OR ALTER PROCEDURE gold.sp_load_kvalita_ovzdusi
AS
BEGIN
    IF OBJECT_ID('gold.fact_kvalita_ovzdusi', 'U') IS NOT NULL
        TRUNCATE TABLE gold.fact_kvalita_ovzdusi;

    -- Agregace per kraj × polutant × hodina
    INSERT INTO gold.fact_kvalita_ovzdusi
    (datum, hodina, kraj_id, kraj_nazev,
     pollutant_kod, pollutant_nazev, jednotka,
     pocet_stanic, prum_value, max_value, min_value,
     pocet_nad_limitem, _gold_loaded_at)
    SELECT
        datum, hodina,
        kraj_id, kraj_nazev,
        pollutant_kod, pollutant_nazev, jednotka,
        COUNT(DISTINCT id_registration) AS pocet_stanic,
        AVG(value)                       AS prum_value,
        MAX(value)                       AS max_value,
        MIN(value)                       AS min_value,
        SUM(CAST(je_nad_limitem AS INT)) AS pocet_nad_limitem,
        SYSDATETIME()
    FROM silver.fact_measurement
    WHERE kraj_id <> -1            -- Skip unmapped stanice v Gold (mají 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

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

EXEC gold.sp_load_kvalita_ovzdusi;
SELECT TOP 20 * FROM gold.fact_kvalita_ovzdusi
ORDER BY pocet_nad_limitem DESC, prum_value DESC;

Talk track: "Gold = star schema fakt. Pro snapshot ~ 14 krajů × 3-10 polutantů × 1 hodina = ~150 řádků. Pro multi-day historii násobeno hodinami × dny — pořád BI-friendly velikost."

Step 6: Volitelně — orchestrace přes pipeline

Pipeline activities:
  1. Stored Procedure: silver.sp_load_fact_measurement
  2. Stored Procedure: gold.sp_load_kvalita_ovzdusi
  3. Send Teams notification "Done"

Talk: "Tohle je production setup. Schedule denně 03:00 (po ranní ingestní pipeline), on-failure email."

Pre-built .pbix obsah

Source: DirectLake na gold.fact_kvalita_ovzdusi. Tím se ukáže Direct Lake feature v Day 3.

5 vizualizací:

1. KPI cards (top of canvas):

   • Aktuální průměr PM10 v ČR (µg/m³)
   • Stanic nad limitem (count)
   • Nejhorší kraj právě teď

2. Mapa krajů (filled map nebo Azure Maps):

   • Color = průměrná hodnota PM10 (nebo vybraný polutant)
   • Tooltip: počet stanic, max value, % nad limitem
   • Slicer pro polutant

3. Stacked bar — Top 10 nejhorších stanic:

   • Y axis: stanice
   • Bars: hodnota polutantu
   • Color: typ stanice (urban / traffic / industrial / background)

4. Line chart — denní trend (pokud máš multi-hour data):

   • X axis: hodina
   • Y axis: hodnota
   • Multi-line: per polutant

5. Compliance ribbon:

   • Bar per kraj
   • Color: zelená/žlutá/červená dle % stanic nad limitem

Slicery: Polutant (PM10/NO2/O3...), Kraj, Datum/hodina.

Talk track:

"Tohle je výsledek. ~470 řádků raw → ~150 řádků agregovaných → tenhle dashboard. Direct Lake mode = report čte přímo z Delta Parquet, žádný import refresh. Když Pipeline každou hodinu doplní data, dashboard je aktuální za sekundy.

Tohle je úplný end-to-end flow Fabricu. Stejný workflow byste použili pro vlastní firemní data — zaměnit URL CHMI za vlastní zdroj, dim_station za vaše entity, a dashboard se sebe-aktualizuje."

• Nezkoušej spark.read.csv() přímo HTTPS — Spark to neumí, je třeba mezikrok přes Files
• Nepoužívej inferSchema=true v Spark — projde data 2× a v Day 1 chce každá vteřina šetřit
• Nezačínej demo Capacity Metrics live. Latence 5–30 min na F2 = uvidíš svůj run zítra. Pre-screenshot.
• Nepřepínej kapacitu během runu. Když je workspace na F2, zůstaň. F64 srovnání = jiný workspace, ne migrace.
• Lektor demuje, účastníci sledují — žádné share screen na tenant účastníků
• NESDÍLEJ cesty k souborům s ID workspaců/lakehouseů — anonymizuj ve slidech

Po kurzu zůstanou tyto reusable artefakty:

1. ovzdusi_demo workspace s prefilled lookups — naklonuj na další zakázky
2. 3 Bronze ingest cesty — funkční šablony pro Notebook/DF Gen2/Pipeline (HTTP source)
3. 2 stored procedures (silver, gold) — copy-paste do dalších DW projektů
4. dim_station.csv (autoritativní z ČHMÚ registru), dim_pollutant.csv, dim_kraj.csv — statické lookup, version-controlled
5. .pbix s Direct Lake — ukázka pro Power BI školení