Runbook: NYC Taxi třemi cestami

Stejný dataset (NYC yellow_tripdata) nahraný 3 způsoby do stejného Lakehouse. Cíl: ukázat rozhodovací matici Notebook vs. Dataflow Gen2 vs. Pipeline + co se děje pod kapotou každé varianty (engine, cost driver, hidden mechanics).

Story arc dema:

1. Setup + společný cíl — jeden Lakehouse, 3 prázdné tabulky
2. Cesta A — Notebook (PySpark) — code-first
3. Cesta B — Dataflow Gen2 — low-code, click-through + M v Advanced Editoru
4. Cesta C — Pipeline (Copy Activity + ForEach) — orchestrace
5. Srovnání + decision matrix — kdy co volit, CU spotřeba, čas

Dataset URL pattern:

https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_{YEAR}-{MONTH:02d}.parquet

Default v runbooku: YEAR=2024, MONTH=1 (~3M řádků, ~50 MB). Multi-month rozšíření v bonus sekci.

1. Workspace NYC_3cesty (F2 nebo trial)
2. Lakehouse: LH_NYC_3cesty
3. Prázdná Files struktura — vytvoří se za běhu:

Files/
├── raw_notebook/
├── raw_dfg2/         (DF Gen2 si vytvoří staging samo)
└── raw_pipeline/

Talk track úvod:

"První cesta — code-first. Notebook, PySpark. Volíte když máte data engineera, potřebujete custom logiku, nebo chcete reusable knihovnu transformací. Nejnižší abstrakce, nejvyšší kontrola, ale taky nejvyšší cognitive load."

A.1 — Notebook setup

1. Workspace → New → Notebook
2. Language: PySpark (Python)
3. Add Lakehouse → Existing → LH_NYC_3cesty → Make default

A.2 — Cell 1: Download (Spark neumí HTTPS přímo)

import urllib.request, os, time

YEAR, MONTH = 2024, 1
URL = f"https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_{YEAR}-{MONTH:02d}.parquet"
LOCAL = f"/lakehouse/default/Files/raw_notebook/yellow_tripdata_{YEAR}_{MONTH:02d}.parquet"

os.makedirs(os.path.dirname(LOCAL), exist_ok=True)

t0 = time.time()
urllib.request.urlretrieve(URL, LOCAL)
print(f"Downloaded {os.path.getsize(LOCAL)/1024/1024:.1f} MB in {time.time()-t0:.1f}s")

Talk track:

• "Spark neumí číst přímo z https://. Driver má jen dbfs:/, abfss://, file:///. Workaround = stáhnout přes Python na driver node → mssparkutils nebo přímo /lakehouse/default/Files/."
• "Tahle úvaha vám ušetří hodinu hledání proč spark.read.parquet('https://...') padá."

A.3 — Cell 2: Read parquet → Delta

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

df = spark.read.parquet(f"Files/raw_notebook/yellow_tripdata_{YEAR}_{MONTH:02d}.parquet")

df = (df
    .withColumn("tpep_pickup_datetime",  col("tpep_pickup_datetime").cast("timestamp"))
    .withColumn("tpep_dropoff_datetime", col("tpep_dropoff_datetime").cast("timestamp"))
    .withColumn("_ingested_at",          current_timestamp())
    .withColumn("_ingest_method",        lit("notebook_pyspark"))
    .withColumn("_source_year",          lit(YEAR))
    .withColumn("_source_month",         lit(MONTH))
)

print(f"Rows: {df.count():,}")
df.printSchema()

A.4 — Cell 3: Save jako Delta tabulka

(df.write
   .format("delta")
   .mode("overwrite")
   .option("overwriteSchema", "true")
   .saveAsTable("trips_notebook"))

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

Verify: Lakehouse Explorer → Tables → Refresh → trips_notebook (~3M řádků).

A.5 — Co se děje pod kapotou (talk track pro lektora)

Klíčové body pro publikum:

• "Spark cold start na F2 = ~90 s. Pure Python notebook by startoval instantně. Pro 50 MB / 3M řádků je Spark overkill — proto pure Python alternativa níže."
• "Spark cost = vCore-hours. F2 má 2 background vCores. 1 minuta běhu = 2 vCore-min = ~33 CU(s)."
• "Delta write zapíše parquet + _delta_log/. Otevřete složku Tables/trips_notebook/ v Lakehouse Explorer → uvidíte i _delta_log adresář s JSON commit logem. To je co dělá Lakehouse 'data lake-house' místo jen 'data lake'."

A.6 — Pure Python alternativa (pokud chceš ukázat "Spark není vždy odpověď")

import pandas as pd, urllib.request
from deltalake import write_deltalake

URL = "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-01.parquet"
df = pd.read_parquet(URL)  # pandas umí HTTP přímo
print(f"Rows: {len(df):,}")

write_deltalake(
    "/lakehouse/default/Tables/trips_notebook_purepython",
    df,
    mode="overwrite",
)

Talk: "Žádný Spark, žádný cold start, žádné vCores. Pro single-node workload pod ~5 GB je tohle rychlejší a levnější. DP-600 je hlavně o Spark/SQL, ale tohle vědět = senior move."

A.7 — Risky body

• Spark cold start (1–2 min na F2) — naplánuj povídání o Spark architecture během čekání
• INT96 timestamps někdy neprojdou SQL endpoint bez explicit .cast("timestamp")
• Files/ složka neexistuje při prvním runu — os.makedirs(..., exist_ok=True) ji vytvoří

Talk track úvod:

"Druhá cesta — low-code. Dataflow Gen2 = Power Query v Fabricu. Volíte když máte analytika z Power BI, který už zná M language. UI je shodné s Power Query Online. Pro NYC parquet přes HTTPS je to technicky funkční, ale ekonomicky nejdražší cesta — uvidíte proč."

B.1 — Dataflow Gen2 setup

1. Workspace → New → Dataflow Gen2
2. Get data → Blank query (NE "Web" wizard — chceme přímo M)
3. View → Advanced editor
4. Vlož níže uvedený M skript

B.2 — M skript (Advanced Editor)

let
    // === PARAMETERS ===
    // V produkci: definuj jako Dataflow parameters (Home → Manage parameters)
    // Pro demo: hardcoded
    pYear        = 2024,
    pStartMonth  = 1,
    pEndMonth    = 1,    // změň na 12 pro celý rok

    // === MONTH LIST ===
    Months       = List.Numbers(pStartMonth, pEndMonth - pStartMonth + 1),
    MonthsTable  = Table.FromList(Months, Splitter.SplitByNothing(), {"Month"}),

    // === FETCH FUNCTION ===
    // KRITICKÉ: Web.Contents s RelativePath, ne s dynamic full URL.
    // Důvod: Fabric service refresh dělá firewall classification na BASE URL.
    // Dynamic base URL → "Web.Contents URL must be a constant" chyba při refresh.
    fnFetchMonth = (year as number, month as number) as table =>
        let
            FileName = "yellow_tripdata_"
                       & Number.ToText(year)
                       & "-"
                       & Text.PadStart(Number.ToText(month), 2, "0")
                       & ".parquet",
            Source = Parquet.Document(
                Web.Contents(
                    "https://d37ci6vzurychx.cloudfront.net",
                    [RelativePath = "trip-data/" & FileName]
                )
            ),
            // Schema drift safety — TLC mění schema mezi roky.
            // 2024+: airport_fee (lowercase a)
            // 2025+: cbd_congestion_fee
            // MissingField.UseNull = neexistující sloupec → NULL místo error
            SafeColumns = Table.SelectColumns(
                Source,
                {"VendorID", "tpep_pickup_datetime", "tpep_dropoff_datetime",
                 "passenger_count", "trip_distance", "RatecodeID",
                 "PULocationID", "DOLocationID", "payment_type",
                 "fare_amount", "tip_amount", "total_amount"},
                MissingField.UseNull
            ),
            WithMeta = Table.AddColumn(
                Table.AddColumn(SafeColumns, "_source_year",  each year,  Int64.Type),
                "_source_month", each month, Int64.Type
            )
        in
            WithMeta,

    // === INVOKE PER MONTH ===
    Invoked  = Table.AddColumn(MonthsTable, "Data", each fnFetchMonth(pYear, [Month])),
    Combined = Table.Combine(Invoked[Data]),

    // === TYPING ===
    Typed = Table.TransformColumnTypes(Combined, {
        {"VendorID",              Int64.Type},
        {"tpep_pickup_datetime",  type datetime},
        {"tpep_dropoff_datetime", type datetime},
        {"passenger_count",       Int64.Type},
        {"trip_distance",         type number},
        {"RatecodeID",            Int64.Type},
        {"PULocationID",          Int64.Type},
        {"DOLocationID",          Int64.Type},
        {"payment_type",          Int64.Type},
        {"fare_amount",           type number},
        {"tip_amount",            type number},
        {"total_amount",          type number}
    }),

    // === LINEAGE STAMP ===
    WithIngestMeta = Table.AddColumn(
        Table.AddColumn(Typed, "_ingest_method", each "dataflow_gen2", type text),
        "_ingested_at", each DateTime.LocalNow(), type datetime
    )
in
    WithIngestMeta

B.3 — Pojmenuj query a nastav destination

1. Pravý panel → query name: trips_dfg2
2. Home → Add data destination → Lakehouse
3. Vyber LH_NYC_3cesty, table trips_dfg2, Replace mode
4. Save & Run (vpravo nahoře "Publish")

B.4 — Co se děje pod kapotou (toto je klíčový talk pro demo)

Toto je sekce, kvůli které celé demo dává smysl. Lze nakreslit na flipchart:

┌──────────────────────────────────────────────────────────────────┐
│                     DATAFLOW GEN2 EXECUTION                       │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│   1. Mashup container startuje (cold start ~10–20 s)             │
│           ↓                                                       │
│   2. M evaluator běží query                                      │
│           ↓                                                       │
│   3. Web.Contents → HTTP GET na cloudfront                       │
│           ↓                                                       │
│   4. Parquet.Document → STREAM celý parquet do memory            │
│           ↓                                                       │
│   5. Table.SelectColumns / TransformColumnTypes — row-by-row     │
│      v M evaluatoru (NE distributed!)                            │
│           ↓                                                       │
│   6. Output → SKRYTÝ Staging Lakehouse (auto-vytvořený)          │
│           ↓                                                       │
│   7. Staging Lakehouse → DESTINATION Lakehouse (copy)            │
│                                                                   │
└──────────────────────────────────────────────────────────────────┘

Klíčové body:

1. Mashup Engine — DF Gen2 spouští M v containeru (stejný engine jako Power Query Online v Power BI Service). Single-threaded per query, žádná distribuce.

2. Skrytý Staging Lakehouse — Při prvním runu Dataflow Gen2 v workspacu Fabric automaticky vytvoří DataflowsStagingLakehouse (viditelný jen přes API, ne v UI). Když má query "Enable staging" zapnuté (default ON pro complex queries), výstup se nejdřív materializuje do staging Delta tabulky, a teprve pak kopíruje do destination.

3. Query folding ZDE NEFUNGUJE — Parquet.Document nad Web.Contents není foldable backend (HTTP není query-able). Znamená to:
   • Celý parquet (50 MB) se stáhne do Mashup containeru
   • Parsuje se v paměti v M evaluatoru
   • Všechny Table.SelectColumns, TransformColumnTypes běží row-by-row v M, ne v zdrojovém systému

Pro 12 měsíců = ~600 MB stahování + in-memory parsing. Toto je hlavní cost driver a důvod proč je tato cesta drahá pro velký objem.

4. Web.Contents RelativePath pattern — Pokud bys napsal:

// ŠPATNĚ — refresh v service selže
Web.Contents("https://d37ci6vzurychx.cloudfront.net/trip-data/" & FileName)

service refresh hodí: "Web.Contents URL must be a constant". Důvod: firewall classification potřebuje statický base URL při design-time. Vždy používej RelativePath parametr pro dynamic části.

5. CU spotřeba — orientačně pro 1 měsíc (změř live v Microsoft Fabric Capacity Metrics App):
   • Mashup execution: ~30–60 CU(s)
   • Staging write: ~10–20 CU(s)
   • Destination copy: ~10–20 CU(s)
   • Total: ~50–100 CU(s) na 1 měsíc (vs. ~10–20 CU(s) pro Pipeline Copy Activity ekvivalent)

B.5 — Risky body

• Web.Contents URL must be a constant — když nepoužiješ RelativePath
• Refresh trvá dlouho — pro 1 měsíc 1–3 min na F2 (vs. ~30 s pipeline)
• Destination configurace bug — když "Save & Run" hodí error, smazat destinaci a znovu nastavit
• MissingField.UseNull je ZÁSADNÍ pro multi-year — bez něj refresh padne, jakmile narazí na měsíc s jiným schema

Talk track úvod:

"Třetí cesta — orchestrace. Pipeline. Volíte když máte složitější flow: stáhnout z více zdrojů, zavolat notebook, podmínit, paralelizovat, poslat email. Pro NYC parquet je to nejlevnější a nejrychlejší cesta — Copy Activity dělá binary passthrough, žádný parsing."

C.1 — Vytvoření pipeline (UI walkthrough)

1. Workspace → New → Data pipeline
2. Název: pl_trips_pipeline
3. Activities → přidej ForEach
4. Uvnitř ForEach → přidej Copy data

Pro demo můžeš pipeline postavit klikáním. Níže je JSON definition pro reference / git / reproducibility.

C.2 — Pipeline JSON (Fabric Data Factory)

{
  "name": "pl_trips_pipeline",
  "objectId": "PIPELINE_GUID",
  "properties": {
    "activities": [
      {
        "name": "Set_MonthList",
        "type": "SetVariable",
        "typeProperties": {
          "variableName": "MonthList",
          "value": {
            "value": "@createArray(\n  json('{\"year\":2024,\"month\":\"01\"}'),\n  json('{\"year\":2024,\"month\":\"02\"}'),\n  json('{\"year\":2024,\"month\":\"03\"}')\n)",
            "type": "Expression"
          }
        }
      },
      {
        "name": "ForEach_Month",
        "type": "ForEach",
        "dependsOn": [
          { "activity": "Set_MonthList", "dependencyConditions": ["Succeeded"] }
        ],
        "typeProperties": {
          "items": {
            "value": "@variables('MonthList')",
            "type": "Expression"
          },
          "isSequential": false,
          "batchCount": 4,
          "activities": [
            {
              "name": "Copy_TripData",
              "type": "Copy",
              "typeProperties": {
                "source": {
                  "type": "ParquetSource",
                  "storeSettings": {
                    "type": "HttpReadSettings",
                    "requestMethod": "GET"
                  },
                  "formatSettings": {
                    "type": "ParquetReadSettings"
                  }
                },
                "sink": {
                  "type": "LakehouseTableSink",
                  "tableActionOption": "Append"
                },
                "enableStaging": false,
                "translator": {
                  "type": "TabularTranslator",
                  "typeConversion": true,
                  "typeConversionSettings": {
                    "allowDataTruncation": true,
                    "treatBooleanAsNumber": false
                  }
                }
              },
              "inputs": [
                {
                  "referenceName": "ds_http_nyc_taxi",
                  "type": "DatasetReference",
                  "parameters": {
                    "year":  { "value": "@item().year",  "type": "Expression" },
                    "month": { "value": "@item().month", "type": "Expression" }
                  }
                }
              ],
              "outputs": [
                {
                  "referenceName": "ds_lakehouse_trips_pipeline",
                  "type": "DatasetReference"
                }
              ]
            }
          ]
        }
      }
    ],
    "variables": {
      "MonthList": { "type": "Array" }
    },
    "parameters": {
      "Year":       { "type": "int",    "defaultValue": 2024 },
      "StartMonth": { "type": "string", "defaultValue": "01" },
      "EndMonth":   { "type": "string", "defaultValue": "01" }
    }
  }
}

C.3 — Datasety (referenced v pipeline)

Fabric Data Factory řeší connection a dataset jinak než ADF — některé věci jsou inline v Copy Activity. Pro HTTP source to v UI vypadá takto:

Source connection (HTTP):

• Connection type: HTTP
• Base URL: https://d37ci6vzurychx.cloudfront.net
• Authentication: Anonymous

Source dataset (parametrized URL):

• Relative URL: @concat('trip-data/yellow_tripdata_', string(pipeline().parameters.Year), '-', item().month, '.parquet')
• File format: Parquet
• Compression: none (parquet má vnitřní kompresi)

Sink (Lakehouse table):

• Workspace: NYC_3cesty
• Lakehouse: LH_NYC_3cesty
• Root folder: Tables
• Table name: trips_pipeline
• Table action: Append (pro multi-month) nebo Overwrite (pro replace)

C.4 — Co se děje pod kapotou

┌──────────────────────────────────────────────────────────────────┐
│                  PIPELINE COPY ACTIVITY EXECUTION                 │
├──────────────────────────────────────────────────────────────────┤
│                                                                   │
│   1. Pipeline orchestrator (mgmt plane) startuje run             │
│           ↓                                                       │
│   2. ForEach planner → spawn 4 paralelní Copy Activity            │
│           ↓                                                       │
│   3. Per Copy: alokace DIU (Data Integration Units) workerů       │
│           ↓                                                       │
│   4. Worker → HTTP GET parquet → STREAM (chunked) → ADLS write   │
│      (binary passthrough, ŽÁDNÝ parsing!)                        │
│           ↓                                                       │
│   5. Sink writer → Delta convert (parquet už máme, jen           │
│      _delta_log/ commit JSON)                                    │
│                                                                   │
└──────────────────────────────────────────────────────────────────┘

Klíčové body:

1. Copy engine = binary passthrough — Copy Activity pro Parquet→Parquet NEPARSUJE data. Stahuje bytes z HTTP a píše bytes na ADLS. Žádný row-by-row, žádný in-memory parsing. Toto je hlavní důvod proč je Pipeline 5–10× levnější než DF Gen2 pro raw landing.

2. DIU (Data Integration Units) — Pipeline cost je v DIU-hours. F2 alokuje typicky 4 DIUs per Copy Activity. 1 měsíc parquet (50 MB) = ~10–20 s Copy → ~0.05 DIU-h → ~10–20 CU(s).

3. ForEach paralelismus — isSequential: false + batchCount: 4 znamená 4 měsíce stahované zároveň. Pro 12 měsíců = 3 vlny × ~20 s = ~60 s total. Vs. DF Gen2, kde by 12 měsíců sériově trvalo ~5+ min.

4. enableStaging: false — Pro HTTP→Lakehouse direct copy nepotřebuješ staging blob. Šetří další cost a IO.

5. tableActionOption: Append — pro multi-month iterativní load. Pro replace mode nastav Overwrite. Pro idempotent re-runs: kombinuj s "Delete activity" před ForEach, který smaže Lakehouse table.

6. CU spotřeba — orientačně pro 1 měsíc:

   • Pipeline orchestration overhead: ~5 CU(s)
   • Copy Activity (DIU): ~10–20 CU(s)
   • Total: ~15–25 CU(s) na 1 měsíc (vs. ~50–100 CU(s) pro DF Gen2 ekvivalent)

C.5 — Risky body

• HTTP source flaky — cloudfront občas vrátí 502/timeout. Přidej retry: 3 na Copy Activity (Settings tab).
• Schema drift v multi-month — Copy Activity s tableActionOption: Append selže, když měsíc má jiné sloupce (např. 2024 vs. 2025 přidá cbd_congestion_fee). Řešení: pro multi-year použij landing do Files (raw parquet bez table convert), pak Notebook na Delta convert s schema merge.
• Fabric DF nemá "Import from JSON" — JSON výše je referenční. Pro postavení v praxi: klikat v UI dle hodnot z JSON, nebo použít Fabric REST API (POST /workspaces/{wsId}/items s definition payload).

C.6 — Schedule trigger

Pipeline → top right → Schedule → On → Daily / Hourly / Custom CRON.

Pro NYC TLC: dataset se updatuje měsíčně, ~2 měsíce zpožděně (2024-12 data dostupná ~2025-02). Realistický schedule = monthly trigger 5. dne v měsíci.

Live measurement v demu (volitelně)

Před demem (24+ h zpět, kvůli Capacity Metrics latency) spusť všechny 3 cesty na stejném dni a hodině. Ráno screenshot z Microsoft Fabric Capacity Metrics App → "Compute" tab → filter podle workspace NYC_3cesty → "Items" view.

Měříš:

• CU(s) per item type (Notebook / Dataflow / Pipeline)
• Duration per run
• Throttling indicator (žádný by neměl být na F2 pro single-month)

Tahle čísla v real-time projekci z účastníkova pohledu jsou nejvíc přesvědčivá ukázka "proč si vybrat tu správnou cestu".

• Nezkoušet spark.read.parquet('https://...') — Spark to neumí, padne. Vždy nejdřív urllib download.
• Nepoužívat dynamic full URL v Web.Contents — refresh v service selže s "URL must be constant". Vždy RelativePath.
• Nezakládat Pipeline JSON ručně přes API v live demu — Fabric REST API je flaky, použij UI walkthrough.
• Neukazovat Capacity Metrics live během dema — latence 5–30 min na F2. Pre-screenshot z předchozího dne.
• Nenechávat Spark session viset — po Notebook části ručně Stop session (pravý horní roh notebooku) jinak žere CU dál.
• Pro multi-month NEPOUŽÍVAT DF Gen2 — předem si změř, kolik CU to sežere, ať máš číslo k vystrašení publika ("12 měsíců přes DF Gen2 = 1200 CU(s) = 20 min F2 capacity uptime").

Cesta A (Notebook):

for MONTH in range(1, 13):
    URL = f"https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-{MONTH:02d}.parquet"
    urllib.request.urlretrieve(URL, f"/lakehouse/default/Files/raw_notebook/yellow_tripdata_2024_{MONTH:02d}.parquet")

df = spark.read.parquet("Files/raw_notebook/yellow_tripdata_2024_*.parquet")  # glob

Cesta B (Dataflow Gen2):

Změň v M skriptu: pEndMonth = 12. Refresh trvá ~5–10 min na F2. Sleduj CU spotřebu — to je ten "wow moment" pro publikum.

Cesta C (Pipeline):

Změň Set_MonthList activity:

@createArray(
  json('{"year":2024,"month":"01"}'),
  json('{"year":2024,"month":"02"}'),
  ...
  json('{"year":2024,"month":"12"}')
)

Nebo elegantněji: pipeline expression generuje array dynamicky:

@range(1, 12)

A v Copy Activity URL: @concat('trip-data/yellow_tripdata_2024-', if(less(item(), 10), concat('0', string(item())), string(item())), '.parquet')

ForEach s batchCount: 4 → 12 měsíců paralelně po 4 → ~60–90 s total na F2.

Po dokončení dema máš:

1. 3 funkční Bronze cesty (Notebook, DF Gen2, Pipeline) — copy-paste do dalších kurzů s jiným datasetem
2. M skript s Web.Contents RelativePath patternem a schema-drift safety — reusable pattern pro jakýkoliv parquet-over-HTTP zdroj
3. Pipeline JSON — referenční šablona pro ForEach + HTTP Copy
4. CU srovnávací tabulka — slide pro "kdy co volit"

Návaznost: Pro Silver/Gold + PBI report nad NYC Taxi data pokračuj v runbooku Mini Medallion NYC Taxi (Cell 3.x dál).