From 3116ca55791a69251269955d52f9c45e6104b773 Mon Sep 17 00:00:00 2001 From: Pawel Bednarski Date: Mon, 20 Jul 2026 09:38:00 +0100 Subject: [PATCH 1/6] feat: rewrite the pipeline as Lakeflow Spark Declarative Pipelines Replace the ingest/transform/aggregate jobs with a single pipeline declaring three datasets. Execution order comes from the table references in the code rather than from running jobs in sequence, and the runtime owns checkpoints, triggers, output modes, and the writes themselves. The transformation logic is untouched: both branches import the same transform_silver and build_gold from the windmill wheel, covered by the same 43 unit tests. Only orchestration differs, which is what makes them comparable. Bronze and silver are streaming tables. Gold is a materialized view rather than a streaming table, because streaming tables never revisit rows they have already emitted -- a daily aggregate built as one would go stale the moment late or corrected data reached silver. It reads silver with spark.read, not readStream: a streaming read would impose watermark and state constraints for no benefit, since the anomaly baseline spans the whole history rather than a window. Every expectation is expect_all -- warn and keep, never _or_drop or _or_fail. That is the declarative form of a decision made on the first branch: invalid readings are flagged, not discarded. Dropping them hides sensor faults from the people who fix them, and failing the update would take the pipeline down over a few bad rows in an otherwise good delivery. Silver's conditions reference the boolean columns the transform already produces instead of restating the rules, so the rule cannot drift between column and expectation. Expectation metrics reproduce the fixture manifest independently: 27 bad turbine ids, 9 bad power readings, 10 bad wind speeds, 8 bad bearings, 20 unknown-group rows -- summing to the manifest's 54 invalid readings. Gold's three expectations pass 465/465. Results match the job branches exactly: silver 11169, 54 invalid, 20 unknown, 465 turbine-days, 20 anomalies, and the stage 1 validation job passes unchanged against pipeline-produced tables. One migration constraint found the hard way: a pipeline cannot adopt an existing managed table. The first run failed because the job branches had written plain Delta tables at those names; they had to be dropped so the pipeline could own them. Harmless for regenerable fixtures, but on a real migration it means a cutover plan rather than a switch. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 61 +++++---- docs/declarative_pipeline.md | 150 +++++++++++++++++++++ notebooks/02_bronze_ingestion.py | 47 ------- notebooks/03_silver_transform.py | 42 ------ notebooks/04_gold_aggregate.py | 48 ------- resources/windmill.pipeline.yml | 34 +++++ resources/windmill_aggregate.job.yml | 28 ---- resources/windmill_ingest.job.yml | 28 ---- resources/windmill_transform.job.yml | 28 ---- transformations/01_bronze_turbine_raw.py | 39 ++++++ transformations/02_silver_turbine_clean.py | 52 +++++++ transformations/03_gold_turbine_summary.py | 42 ++++++ 12 files changed, 350 insertions(+), 249 deletions(-) create mode 100644 docs/declarative_pipeline.md delete mode 100644 notebooks/02_bronze_ingestion.py delete mode 100644 notebooks/03_silver_transform.py delete mode 100644 notebooks/04_gold_aggregate.py create mode 100644 resources/windmill.pipeline.yml delete mode 100644 resources/windmill_aggregate.job.yml delete mode 100644 resources/windmill_ingest.job.yml delete mode 100644 resources/windmill_transform.job.yml create mode 100644 transformations/01_bronze_turbine_raw.py create mode 100644 transformations/02_silver_turbine_clean.py create mode 100644 transformations/03_gold_turbine_summary.py diff --git a/README.md b/README.md index 8439b31..013c841 100644 --- a/README.md +++ b/README.md @@ -46,16 +46,9 @@ databricks bundle deploy --target dev databricks bundle run --target dev windmill_init ``` -6. **Run data pipeline** +6. **Run the pipeline** (bronze, silver and gold in one update) ```bash -# Ingest raw data -databricks bundle run --target dev windmill_ingest - -# Transform data -databricks bundle run --target dev windmill_transform - -# Aggregate to business layer -databricks bundle run --target dev windmill_aggregate +databricks bundle run --target dev windmill_pipeline ``` ## Architecture @@ -64,13 +57,17 @@ databricks bundle run --target dev windmill_aggregate ``` data_group_*.csv (Raw) ↓ -[01_copy_data] → Volume: dropzone - ↓ -[02_bronze_ingestion] → Table: bronze.turbine_raw +[windmill_init] → Volume: dropzone/inbox ↓ -[03_silver_transform] → Table: silver.turbine_clean - ↓ -[04_gold_aggregate] → Table: gold.turbine_summary +┌─ windmill_pipeline ─────────────────────────────────┐ +│ turbine_raw (streaming table, Auto Loader) │ +│ ↓ │ +│ turbine_clean (streaming table, + expectations) │ +│ ↓ │ +│ turbine_summary (materialized view, aggregate) │ +└─────────────────────────────────────────────────────┘ + +Order is derived from the table references in the code, not declared. ``` ### Catalog Structure @@ -95,7 +92,9 @@ timestamps). Files are archived after ingestion via `cloudFiles.cleanSource`. Append-only: nothing is filtered or corrected here. ### Silver Layer -Cleaning and row-level validation, read incrementally from bronze as a stream. +A streaming table, read incrementally from bronze. Every check is declared as an +`expect_all` expectation — warn and keep, never drop or fail — so violations are +counted per run in the event log while the rows themselves pass through. - Deduplication on `(timestamp, turbine_id)` via `dropDuplicatesWithinWatermark`, bounded by a 2-day watermark so streaming state cannot grow without limit @@ -112,7 +111,9 @@ outlier detection is deliberately *not* here — it needs a population, and the watermark exists to bound dedup state, not to define a statistical window. ### Gold Layer -Batch aggregation and anomaly detection over valid readings only. +A materialized view, not a streaming table: streaming tables never revisit rows +they have emitted, so an aggregate built as one would go stale the moment late +data landed. Aggregation and anomaly detection over valid readings only. - Daily min / max / mean / stddev of `power_output` per turbine - Per-turbine baseline computed over **daily means**, so the unit being tested @@ -140,11 +141,13 @@ windmill_classcell/ │ ├── silver.schema.yml │ ├── gold.schema.yml │ ├── dropzone.volume.yml +│ ├── windmill.pipeline.yml │ ├── windmill_init.job.yml -│ ├── windmill_ingest.job.yml -│ ├── windmill_transform.job.yml -│ ├── windmill_aggregate.job.yml │ └── windmill_validate.job.yml +├── transformations/ # Pipeline datasets, one per file +│ ├── 01_bronze_turbine_raw.py +│ ├── 02_silver_turbine_clean.py +│ └── 03_gold_turbine_summary.py ├── README.md # This file ├── .gitignore ├── data/ # CSV data files @@ -152,16 +155,14 @@ windmill_classcell/ │ ├── data_group_2.csv │ └── data_group_3.csv ├── data_test/ # Corrupted fixtures + expectation manifest (generated) -├── notebooks/ # Databricks notebooks -│ ├── 01_copy_data.py -│ ├── 02_bronze_ingestion.py -│ ├── 03_silver_transform.py -│ ├── 04_gold_aggregate.py +├── notebooks/ # Jobs that sit outside the pipeline +│ ├── 01_copy_data.py # Seeds the dropzone volume │ └── 05_validate.py # Asserts pipeline output against the manifest ├── scripts/ │ └── generate_test_data.py └── docs/ ├── architecture.md + ├── declarative_pipeline.md ├── testing.md └── dev_seed_original_results.md ``` @@ -226,12 +227,16 @@ known expected outcomes, and asserts against them. python scripts/generate_test_data.py # deterministic; regenerates data_test/ databricks bundle deploy --target test databricks bundle run --target test windmill_init -databricks bundle run --target test windmill_ingest -databricks bundle run --target test windmill_transform -databricks bundle run --target test windmill_aggregate +databricks bundle run --target test windmill_pipeline databricks bundle run --target test windmill_validate # fails the job on any violation ``` +Expectations and the validation job cover different failures and neither replaces +the other. Expectations report what the pipeline saw, per run; the validation job +asserts what it *should* have seen, against a manifest generated independently. +An expectation cannot notice that a whole file was skipped, nor that silver +stayed invariant across repeated loads. + Covers invalid turbine IDs, negative and null measurements, out-of-range wind direction, statistical outliers, missing rows, duplicates, unmatched filenames, and boundary values that must stay valid. Writes to `test_bronze` / `test_silver` diff --git a/docs/declarative_pipeline.md b/docs/declarative_pipeline.md new file mode 100644 index 0000000..23de9b8 --- /dev/null +++ b/docs/declarative_pipeline.md @@ -0,0 +1,150 @@ +# Declarative rewrite (Lakeflow Spark Declarative Pipelines) + +The job-based branches orchestrate three notebooks that each read a table, write +a table, and manage their own checkpoint. This branch replaces them with one +pipeline that declares the three datasets and lets the runtime work out the rest. + +The transformation logic is unchanged — both branches import the same +`transform_silver` and `build_gold` from the `windmill` wheel, covered by the +same 43 unit tests. Only the orchestration differs, which is what makes the two +comparable. + +## What the rewrite removes + +| Concern | Job-based branches | This branch | +|---|---|---| +| Execution order | Three jobs run in sequence, by hand or by a scheduler | Derived from table references in the code | +| Checkpoints | Explicit path per stream, hand-placed in a volume | Managed by the pipeline | +| Triggers / output modes | `trigger(availableNow=True)`, `outputMode("append")` per write | Implicit in the dataset type | +| Writes | `saveAsTable` / `toTable`, plus `overwriteSchema` when columns change | The decorator's return value is the table | +| Table type | Everything is a plain managed Delta table | Streaming tables for append, materialized view for the aggregate | +| Quality reporting | A separate job asserting counts after the fact | Expectations recorded per run in the event log | + +Three job YAML files and three notebooks are deleted; three transformation files +and one pipeline YAML replace them. + +## Dataset types + +Bronze and silver are **streaming tables**: both are append-only, and both +process only what arrived since the last run. + +Gold is a **materialized view**, deliberately. Streaming tables never revisit +rows they have already emitted, so a daily aggregate computed as one would go +stale as soon as late or corrected data landed in silver. A materialized view +recomputes. The read is `spark.read.table`, not `readStream` — reading silver as +a stream would impose watermark and state constraints for no benefit, since the +anomaly baseline is computed over the whole history rather than a window. + +## Expectations + +Every expectation is `expect_all` — warn and keep. Never `_or_drop`, never +`_or_fail`. That is the declarative statement of a decision made on the first +branch: invalid readings are flagged, not discarded. Dropping them would hide +sensor faults from the people who have to fix them, and failing the update would +take the pipeline down over a handful of bad rows in an otherwise good delivery. + +The silver conditions reference the boolean columns the transform already +produces (`is_power_output_valid`, and so on) rather than restating the rules in +SQL. One source of truth: the rule cannot drift between the column and the +expectation, and the column is unit tested in the package. + +### Measured against the corrupted fixtures + +| Expectation | Failed | Manifest expects | +|---|---:|---:| +| `turbine_id_in_expected_group` | 27 | 27 | +| `power_output_present_and_non_negative` | 9 | 9 | +| `wind_speed_present_and_non_negative` | 10 | 10 | +| `wind_direction_within_compass_range` | 8 | 8 | +| `source_file_maps_to_known_group` | 20 | 20 | +| `timestamp_present` | 0 | 0 | +| `turbine_id_present` | 0 | 0 | + +27 + 9 + 10 + 8 = 54, the manifest's `invalid_readings_total`. The expectations +and the fixture generator arrive at the same numbers by completely independent +routes, which is a stronger check than either alone. + +Gold's three expectations pass 465/465: every turbine-day has readings, every +anomaly flag is decided rather than null, and no day contains sub-hourly +readings. + +Reading the metrics back: + +```sql +SELECT explode(from_json( + details:flow_progress.data_quality.expectations, + 'array>' + )) AS e +FROM event_log("") +WHERE details:flow_progress.data_quality.expectations IS NOT NULL +``` + +Note the CLI's `list-pipeline-events --output json` flattens nested details and +returns `flow_progress` as an empty object. Query the event log rather than the +CLI for anything nested. + +## Results + +Identical to the job-based branches on every layer that matters: + +| Metric | Jobs | Pipeline | +|---|---:|---:| +| silver rows | 11,169 | 11,169 | +| invalid readings | 54 | 54 | +| `UNKNOWN` group rows | 20 | 20 | +| turbine-days | 465 | 465 | +| anomalies | 20 | 20 | + +The stage 1 validation job passes unchanged against pipeline-produced tables — +it queries the same table names and asserts the same manifest, so it is a fair +cross-check rather than a rewritten test. + +Bronze reads 33,534 because the inbox still held three loads when the pipeline +first ran: `cloudFiles.cleanSource` archives on a retention delay, so previously +ingested files had not yet moved, and the pipeline's checkpoint was new. Dedup +collapsed them to the same 11,169 silver rows, which is the invariant that +matters. + +## Migration constraint worth knowing + +**A pipeline cannot adopt an existing managed table.** The first run failed with: + +``` +Could not materialize `...`.`turbine_raw` because a MANAGED table already +exists with that name. +``` + +The job-based branches had written plain Delta tables at those names. Converting +them to pipeline-managed datasets is not an in-place operation — the tables must +be dropped so the pipeline can create and own them. Fine here, where the data is +regenerable fixtures, but on a real migration it means a cutover plan: either +write the pipeline to new names and swap, or accept a rebuild window. + +## What this branch keeps + +`windmill_init` still seeds the dropzone, because getting files into the volume +is not the pipeline's job. + +`windmill_validate` still runs, and is still worth having. Expectations report +what the pipeline saw; the validation job asserts what it *should* have seen, +against a manifest generated independently of the pipeline. They fail on +different things — an expectation cannot notice that a whole file was skipped, +and it cannot check that silver stayed invariant across repeated loads. + +## Trade-offs + +**In favour.** Much less orchestration code. The dependency graph is derived +rather than maintained. Data quality is a first-class, per-run, queryable +artifact instead of a bespoke job. Table types force an explicit answer to +"should this recompute or append?". + +**Against.** The pipeline owns its tables, so migration is a cutover, not a +switch. Debugging moves from reading a notebook top to bottom to reading the +pipeline graph and the event log. Failures surface as flow errors, one level +removed from the code that caused them. And the whole thing is Databricks +specific in a way the job-based version — plain PySpark plus a scheduler — is +not. + +Neither is strictly better. The declarative version is the stronger choice when +the pipeline shape is stable and quality reporting matters; the job version is +easier to reason about when the logic is still moving, and it ports. diff --git a/notebooks/02_bronze_ingestion.py b/notebooks/02_bronze_ingestion.py deleted file mode 100644 index 51f7a7b..0000000 --- a/notebooks/02_bronze_ingestion.py +++ /dev/null @@ -1,47 +0,0 @@ -# Databricks notebook source -# Bronze layer: Stream raw turbine data with Auto Loader (cloudFiles) - -from pyspark.sql import SparkSession -from pyspark.sql.functions import current_timestamp, col - -from windmill.schema import TURBINE_SCHEMA - -catalog_name = dbutils.widgets.get("catalog_name") -schema_name = dbutils.widgets.get("schema_name") - -spark = SparkSession.builder.appName("bronze_ingestion").getOrCreate() - -volume_path = f"/Volumes/{catalog_name}/{schema_name}/dropzone/inbox" -archive_path = f"/Volumes/{catalog_name}/{schema_name}/dropzone/archive" - -# Checkpoint kept out of the scan path; pathGlobFilter restricts to CSVs -checkpoint_path = f"/Volumes/{catalog_name}/{schema_name}/dropzone/_checkpoints/turbine_raw" -table_name = f"{catalog_name}.{schema_name}.turbine_raw" - -# Read: Auto Loader options only -df = ( - spark.readStream - .format("cloudFiles") - .option("cloudFiles.format", "csv") - .option("header", "true") - .option("pathGlobFilter", "*.csv") - .option("cloudFiles.cleanSource", "MOVE") - .option("cloudFiles.cleanSource.moveDestination", archive_path) - .option("cloudFiles.cleanSource.retentionDuration", "7 days") - .schema(TURBINE_SCHEMA) - .load(volume_path) - .withColumn("ingestion_timestamp", current_timestamp()) - .withColumn("source_file", col("_metadata.file_path")) - .withColumn("source_file_timestamp", col("_metadata.file_modification_time")) -) - -# Write: sink options here (checkpoint, trigger, output mode) -( - df.writeStream - .option("checkpointLocation", checkpoint_path) - .outputMode("append") - .trigger(availableNow=True) # run-once batch; only trigger supported on serverless - .toTable(table_name) -) - -print(f"✓ Streamed to {table_name}") diff --git a/notebooks/03_silver_transform.py b/notebooks/03_silver_transform.py deleted file mode 100644 index 2d0406d..0000000 --- a/notebooks/03_silver_transform.py +++ /dev/null @@ -1,42 +0,0 @@ -# Databricks notebook source -# Silver layer: clean and enrich turbine data incrementally from bronze. -# -# Orchestration only -- read, transform, write. The transform itself lives in -# the windmill package (installed as a wheel by the bundle) so it can be unit -# tested against small local DataFrames without a cluster. - -from pyspark.sql import SparkSession - -from windmill.validation import transform_silver - -dbutils.widgets.text("catalog_name", "windmill_classcell") -dbutils.widgets.text("bronze_schema", "bronze") -dbutils.widgets.text("silver_schema", "silver") - -catalog = dbutils.widgets.get("catalog_name") -bronze_schema = dbutils.widgets.get("bronze_schema") -silver_schema = dbutils.widgets.get("silver_schema") - -spark = SparkSession.builder.appName("silver_transform").getOrCreate() - -bronze_table = f"{catalog}.{bronze_schema}.turbine_raw" -silver_table = f"{catalog}.{silver_schema}.turbine_clean" -# Checkpoint lives in the bronze dropzone volume -- the only volume the bundle -# provisions. -checkpoint_path = ( - f"/Volumes/{catalog}/{bronze_schema}/dropzone/_checkpoints/silver_turbine_clean" -) - -df = spark.readStream.table(bronze_table) - -silver = transform_silver(df) - -( - silver.writeStream.option("checkpointLocation", checkpoint_path) - .outputMode("append") - .trigger(availableNow=True) - .toTable(silver_table) - .awaitTermination() -) - -print(f"✓ Streamed to {silver_table}") diff --git a/notebooks/04_gold_aggregate.py b/notebooks/04_gold_aggregate.py deleted file mode 100644 index caf9ea2..0000000 --- a/notebooks/04_gold_aggregate.py +++ /dev/null @@ -1,48 +0,0 @@ -# Databricks notebook source -# Gold layer: daily summary statistics per turbine plus anomaly detection. -# -# Orchestration only -- read, transform, write. The aggregation logic lives in -# the windmill package so the statistical behaviour can be unit tested; see -# tests/test_aggregation.py, in particular the regression guard on computing -# baselines from daily means rather than raw readings. - -from pyspark.sql import SparkSession -from pyspark.sql.functions import col - -from windmill.aggregation import build_gold - -dbutils.widgets.text("catalog_name", "windmill_classcell") -dbutils.widgets.text("silver_schema", "silver") -dbutils.widgets.text("gold_schema", "gold") - -catalog = dbutils.widgets.get("catalog_name") -silver_schema = dbutils.widgets.get("silver_schema") -gold_schema = dbutils.widgets.get("gold_schema") - -spark = SparkSession.builder.appName("gold_aggregate").getOrCreate() - -silver_table = f"{catalog}.{silver_schema}.turbine_clean" -gold_table = f"{catalog}.{gold_schema}.turbine_summary" - -silver = spark.table(silver_table) -print(f"Read {silver.count()} rows from {silver_table}") - -summary = build_gold(silver) - -# overwriteSchema so adding or removing a summary column does not require -# dropping the table by hand. Safe here because gold is fully derived from -# silver -- every run rebuilds it from scratch. -summary.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable(gold_table) -print(f"✓ Wrote {summary.count()} turbine-days to {gold_table}") - -anomalies = summary.filter(col("is_anomaly")) -print(f"\nAnomalous turbine-days: {anomalies.count()}") -anomalies.select( - "date", - "turbine_id", - "avg_power_output", - "baseline_mean", - "anomaly_lower_bound", - "anomaly_upper_bound", - "anomalous_reading_count", -).orderBy("turbine_id", "date").show(20, truncate=False) diff --git a/resources/windmill.pipeline.yml b/resources/windmill.pipeline.yml new file mode 100644 index 0000000..570b703 --- /dev/null +++ b/resources/windmill.pipeline.yml @@ -0,0 +1,34 @@ +# Declarative replacement for the windmill_ingest / windmill_transform / +# windmill_aggregate jobs. One pipeline owns bronze, silver and gold; the +# dependency graph between them is derived from the table references in the +# transformation files rather than declared by running jobs in an order. +# +# The transformation code imports from the windmill wheel, so the statistical +# logic and its unit tests are shared with the job-based branches unchanged. +resources: + pipelines: + windmill_pipeline: + name: windmill_pipeline + serverless: true + # Triggered rather than continuous: the source is a daily file drop, so + # there is nothing to gain from holding compute open between deliveries. + continuous: false + # Pipeline default target. Datasets override it with fully-qualified names + # so each medallion layer keeps its own schema, matching the job branches. + catalog: ${var.catalog_name} + schema: ${resources.schemas.bronze.name} + libraries: + - glob: + include: ../transformations/** + environment: + dependencies: + # Built by the artifacts block in databricks.yml. + - ../dist/*.whl + configuration: + # Read in the transformation files via spark.conf.get. Schema names are + # target-dependent (dev_p17b_* vs test_p17b_*), so they cannot be + # hardcoded in the source. + windmill.catalog: ${var.catalog_name} + windmill.bronze_schema: ${resources.schemas.bronze.name} + windmill.silver_schema: ${resources.schemas.silver.name} + windmill.gold_schema: ${resources.schemas.gold.name} diff --git a/resources/windmill_aggregate.job.yml b/resources/windmill_aggregate.job.yml deleted file mode 100644 index 8b8db12..0000000 --- a/resources/windmill_aggregate.job.yml +++ /dev/null @@ -1,28 +0,0 @@ -# Gold: daily per-turbine statistics and anomaly detection. -# -# Batch rather than streaming: anomaly detection needs a population, which a -# watermarked stream cannot provide. Overwrite semantics, so re-running is safe. -resources: - jobs: - windmill_aggregate: - name: windmill_aggregate - description: "Aggregate data to gold layer for analysis" - # Serverless rejects a task-level `libraries` field -- dependencies are - # declared as an environment and referenced by key from the task. - environments: - - environment_key: default - spec: - client: "3" - dependencies: - # Built by the artifacts block in databricks.yml. - - ../dist/*.whl - tasks: - - task_key: gold_aggregate - notebook_task: - notebook_path: ../notebooks/04_gold_aggregate.py - base_parameters: - catalog_name: ${var.catalog_name} - silver_schema: ${resources.schemas.silver.name} - gold_schema: ${resources.schemas.gold.name} - environment_key: default - timeout_seconds: 3600 diff --git a/resources/windmill_ingest.job.yml b/resources/windmill_ingest.job.yml deleted file mode 100644 index f37262b..0000000 --- a/resources/windmill_ingest.job.yml +++ /dev/null @@ -1,28 +0,0 @@ -# Bronze: Auto Loader streams CSVs from the dropzone into turbine_raw. -# -# Append-only, explicit schema, run-once trigger. Files are archived after -# ingestion via cloudFiles.cleanSource, and the checkpoint makes re-runs -# incremental rather than reprocessing everything. -resources: - jobs: - windmill_ingest: - name: windmill_ingest - description: "Ingest raw turbine data to bronze layer" - # Serverless rejects a task-level `libraries` field -- dependencies are - # declared as an environment and referenced by key from the task. - environments: - - environment_key: default - spec: - client: "3" - dependencies: - # Built by the artifacts block in databricks.yml. - - ../dist/*.whl - tasks: - - task_key: bronze_load - notebook_task: - notebook_path: ../notebooks/02_bronze_ingestion.py - base_parameters: - catalog_name: ${var.catalog_name} - schema_name: ${resources.schemas.bronze.name} - environment_key: default - timeout_seconds: 3600 diff --git a/resources/windmill_transform.job.yml b/resources/windmill_transform.job.yml deleted file mode 100644 index b054496..0000000 --- a/resources/windmill_transform.job.yml +++ /dev/null @@ -1,28 +0,0 @@ -# Silver: incremental read from bronze, dedup, and row-level validation. -# -# Reads both bronze and silver schema names because it is a stream from one into -# the other. -resources: - jobs: - windmill_transform: - name: windmill_transform - description: "Transform and clean data to silver layer" - # Serverless rejects a task-level `libraries` field -- dependencies are - # declared as an environment and referenced by key from the task. - environments: - - environment_key: default - spec: - client: "3" - dependencies: - # Built by the artifacts block in databricks.yml. - - ../dist/*.whl - tasks: - - task_key: silver_transform - notebook_task: - notebook_path: ../notebooks/03_silver_transform.py - base_parameters: - catalog_name: ${var.catalog_name} - bronze_schema: ${resources.schemas.bronze.name} - silver_schema: ${resources.schemas.silver.name} - environment_key: default - timeout_seconds: 3600 diff --git a/transformations/01_bronze_turbine_raw.py b/transformations/01_bronze_turbine_raw.py new file mode 100644 index 0000000..b38f64b --- /dev/null +++ b/transformations/01_bronze_turbine_raw.py @@ -0,0 +1,39 @@ +# Bronze: raw turbine readings ingested from the dropzone volume. +# +# Declarative equivalent of notebooks/02_bronze_ingestion.py. The Auto Loader +# read is unchanged; what disappears is the checkpoint path, the trigger, the +# output mode, and the write itself -- the pipeline owns all of that. + +from pyspark import pipelines as dp +from pyspark.sql.functions import col, current_timestamp + +from windmill.schema import TURBINE_SCHEMA + +catalog = spark.conf.get("windmill.catalog") +bronze_schema = spark.conf.get("windmill.bronze_schema") + +VOLUME_ROOT = f"/Volumes/{catalog}/{bronze_schema}/dropzone" + + +@dp.table( + name=f"{catalog}.{bronze_schema}.turbine_raw", + comment="Raw turbine readings as delivered, with ingestion metadata.", +) +def turbine_raw(): + return ( + spark.readStream.format("cloudFiles") + .option("cloudFiles.format", "csv") + .option("header", "true") + .option("pathGlobFilter", "*.csv") + # Archive files once ingested so the inbox reflects what is still pending. + .option("cloudFiles.cleanSource", "MOVE") + .option("cloudFiles.cleanSource.moveDestination", f"{VOLUME_ROOT}/archive") + .option("cloudFiles.cleanSource.retentionDuration", "7 days") + # Explicit schema rather than inference: a malformed file must not + # silently change column types between runs. + .schema(TURBINE_SCHEMA) + .load(f"{VOLUME_ROOT}/inbox") + .withColumn("ingestion_timestamp", current_timestamp()) + .withColumn("source_file", col("_metadata.file_path")) + .withColumn("source_file_timestamp", col("_metadata.file_modification_time")) + ) diff --git a/transformations/02_silver_turbine_clean.py b/transformations/02_silver_turbine_clean.py new file mode 100644 index 0000000..d2b279f --- /dev/null +++ b/transformations/02_silver_turbine_clean.py @@ -0,0 +1,52 @@ +# Silver: deduplicated, validated readings. +# +# Declarative equivalent of notebooks/03_silver_transform.py. The transform is +# the same `transform_silver` from the windmill wheel that the job-based branch +# calls and that the unit tests exercise -- the pipeline changes how it is +# orchestrated, not what it computes. +# +# Expectations here are all `expect_all` (warn), never `_or_drop` or `_or_fail`, +# which is the declarative expression of a decision made on the first branch: +# invalid readings are flagged and kept, not discarded. Dropping them would hide +# sensor faults from the people who need to fix them, and `_or_fail` would take +# the pipeline down over a handful of bad rows in an otherwise good delivery. +# +# The conditions reference the boolean columns produced by the transform rather +# than restating the rules. One source of truth: the rule cannot drift between +# the column and the expectation, and each is unit tested in the package. + +from pyspark import pipelines as dp + +from windmill.validation import transform_silver + +catalog = spark.conf.get("windmill.catalog") +bronze_schema = spark.conf.get("windmill.bronze_schema") +silver_schema = spark.conf.get("windmill.silver_schema") + + +@dp.table( + name=f"{catalog}.{silver_schema}.turbine_clean", + comment="Deduplicated readings with per-field validity flags.", +) +@dp.expect_all( + { + # Row-level validity. Violations are counted and surfaced in the + # pipeline's data quality view while the rows themselves pass through. + "turbine_id_in_expected_group": "is_turbine_id_valid", + "power_output_present_and_non_negative": "is_power_output_valid", + "wind_speed_present_and_non_negative": "is_wind_speed_valid", + "wind_direction_within_compass_range": "is_wind_direction_valid", + # Structural expectations that have no corresponding column: a reading + # without a timestamp or turbine cannot be deduplicated or attributed, + # so it is worth watching separately from the value checks. + "timestamp_present": "timestamp IS NOT NULL", + "turbine_id_present": "turbine_id IS NOT NULL", + # The source filename should identify a turbine group. An UNKNOWN here + # means a file arrived that nobody planned for. + "source_file_maps_to_known_group": "turbine_group != 'UNKNOWN'", + } +) +def turbine_clean(): + return transform_silver( + spark.readStream.table(f"{catalog}.{bronze_schema}.turbine_raw") + ) diff --git a/transformations/03_gold_turbine_summary.py b/transformations/03_gold_turbine_summary.py new file mode 100644 index 0000000..3fa6f90 --- /dev/null +++ b/transformations/03_gold_turbine_summary.py @@ -0,0 +1,42 @@ +# Gold: daily per-turbine statistics and anomaly flags. +# +# A materialized view, not a streaming table. Streaming tables are append-only +# and never revisit rows they have already emitted, so an aggregate computed as +# one would go stale the moment late or corrected data landed in silver. A +# materialized view recomputes, which is what an aggregation needs. +# +# Note the batch read -- `spark.read.table`, not `readStream`. Reading silver as +# a stream here would make this a streaming aggregation with all the watermark +# and state constraints that implies, for no benefit: the anomaly baseline is +# deliberately computed over the whole history, not a window. + +from pyspark import pipelines as dp + +from windmill.aggregation import build_gold + +catalog = spark.conf.get("windmill.catalog") +silver_schema = spark.conf.get("windmill.silver_schema") +gold_schema = spark.conf.get("windmill.gold_schema") + + +@dp.materialized_view( + name=f"{catalog}.{gold_schema}.turbine_summary", + comment="Daily min/max/mean power per turbine, with anomaly flags.", +) +@dp.expect_all( + { + # An anomaly flag that is null silently disappears from any + # `WHERE is_anomaly` filter downstream, so a null here is a defect in + # the aggregation rather than a data quality problem in the source. + "anomaly_flag_is_decided": "is_anomaly IS NOT NULL", + # A turbine-day is built from readings, so it should never be empty. + "day_has_readings": "measurement_count > 0", + # The pipeline assumes one reading per turbine per hour. This does not + # break anything loudly when violated -- it quietly reweights the daily + # mean toward whichever hours sent more samples -- so it is watched + # explicitly. See the assumption note in the README. + "one_reading_per_hour": "NOT has_sub_hourly_readings", + } +) +def turbine_summary(): + return build_gold(spark.read.table(f"{catalog}.{silver_schema}.turbine_clean")) From 28c2a702c7aa1d093903d1e808ee1e53b1718dd3 Mon Sep 17 00:00:00 2001 From: Pawel Bednarski Date: Mon, 20 Jul 2026 10:54:59 +0100 Subject: [PATCH 2/6] docs: document environment isolation, WAP, and the branch comparison Add three sections to the README. Implementation branches -- what each of the three stages adds, a comparison across orchestration, testability, quality reporting, table ownership, migration cost and portability, and the note that all three produce identical results with the stage 1 validation job passing unchanged on each. That last part is what makes the comparison fair rather than three separate claims. Environments and isolation -- how one name_prefix line per target separates schemas as well as job names, why that makes end-to-end testing safe (a service principal can exercise ingestion through assertions with no blast radius on real data), and how a preprod target for full-volume performance and quality testing is a target block rather than an architecture change. Write-Audit-Publish -- the pattern the target isolation enables. Write and Audit are implemented here; Publish is explicitly not, since there is no production target to promote into and building one would be scaffolding around a hypothetical. The shape a promotion step would take is described instead. Also records why the two audit mechanisms are kept independent: expectations report what the pipeline saw, the validation job asserts what it should have seen against an independently generated manifest, and neither can catch the other's failures. Switching between branches documents that a destroy is required first, and why: a pipeline cannot adopt an existing managed table, so job-written tables must be dropped before the pipeline can own them. Harmless for regenerable fixtures, a cutover plan on a real migration. Note the pattern's established name is Write-Audit-Publish, not Write-Ahead- Publish -- the audit gate is the middle step, and it is the one usually skipped. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 178 ++++++++++++++++++++++- transformations/01_bronze_turbine_raw.py | 1 - 2 files changed, 175 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 013c841..2e0bcdd 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,90 @@ -# Windmill - Renewable Energy Data Pipeline +# Windmill — turbine data pipeline + +Ingests raw turbine measurements delivered as daily CSVs, cleans and validates +them, computes per-turbine summary statistics, and identifies turbines whose +output has deviated from expectation. + +Databricks, medallion architecture (bronze → silver → gold), deployed as a +Declarative Automation Bundle. This branch implements the pipeline with +**Lakeflow Spark Declarative Pipelines**; two alternative implementations live on +other branches — see [Implementation branches](#implementation-branches). + +## Implementation branches + +The same problem is solved three ways, so each design decision is visible as a +reviewable diff rather than asserted in prose. + +| Branch | Orchestration | Logic lives in | Data quality | +|---|---|---|---| +| [`medallion_workflow`](../../tree/medallion_workflow) | Three jobs run in sequence | Inline in notebooks | A job asserting counts after the fact | +| [`medallion_whl`](../../tree/medallion_whl) | Three jobs run in sequence | A Python wheel, unit tested | Same, plus 43 unit tests | +| **`medallion_sdp`** (this branch, and `main`) | One declarative pipeline | The same wheel | Expectations recorded per run, plus the assertion job | + +### What each stage adds + +**`medallion_workflow`** — the working pipeline. Auto Loader ingestion, streaming +silver with watermarked dedup, batch gold with anomaly detection. Introduces the +deterministic corrupted-data fixture generator and the validation job that +asserts against it. + +**`medallion_whl`** — extracts the transformation logic into `src/windmill/`, +built as a wheel by the bundle and attached to jobs. Notebooks become +orchestration only. The logic gains unit tests that run against small local +DataFrames in ~11 seconds with no cluster — a loop fast enough to actually catch +statistical bugs, which the integration harness is too slow to do. + +**`medallion_sdp`** — replaces the three jobs with one pipeline. Execution order +is derived from table references rather than declared; the runtime owns +checkpoints, triggers, output modes and writes. Row-level checks become +expectations, so quality is a queryable per-run artifact instead of a bespoke job. + +### Comparison + +| | `medallion_workflow` | `medallion_whl` | `medallion_sdp` | +|---|---|---|---| +| Lines of orchestration | Most | Same | Least | +| Execution order | Declared by running jobs in sequence | Same | Derived from table references | +| Checkpoints, triggers, writes | Hand-managed per stream | Same | Owned by the runtime | +| Unit testable without a cluster | No | **Yes** | **Yes** | +| Quality metrics per run | No | No | **Yes**, in the event log | +| Table ownership | Plain managed Delta | Same | Pipeline-owned | +| Migration in / out | Free | Free | **Cutover required** | +| Portability | Plain PySpark + any scheduler | Same | Databricks-specific | + +None is strictly best. `medallion_whl` is the strongest general engineering +artifact — packaged, tested, portable. `medallion_sdp` is the better choice when +the pipeline shape is stable and quality reporting matters, at the cost of being +tied to the platform and harder to migrate away from. `medallion_workflow` is the +easiest to reason about while the logic is still moving. + +All three produce **identical results** on the same input: silver 11,169 rows, 54 +invalid readings, 465 turbine-days, 20 anomalies. The stage 1 validation job +passes unchanged on all three, which is what makes the comparison fair. + +### Switching between branches + +Branches are not drop-in replacements for each other **in a workspace that +already has a deployment**. Destroy first: -A Databricks demo project showcasing a medallion architecture (bronze/silver/gold) for processing turbine power generation data. +```bash +databricks bundle destroy --target dev # and --target test +git checkout +databricks bundle deploy --target dev +``` + +The reason is table ownership. The job branches write plain managed Delta tables; +the pipeline branch needs to create and own its datasets. A pipeline **cannot +adopt an existing managed table** — it fails with: + +``` +Could not materialize `...`.`turbine_raw` because a MANAGED table already +exists with that name. +``` + +Switching the other way (pipeline → jobs) has the mirror problem. For this +repository that is a non-issue: everything is regenerable from the CSVs in one +command. On a real migration it means a **cutover plan** — write to new names and +swap, or accept a rebuild window. ## Quick Start @@ -217,6 +301,90 @@ windmill_classcell/ 7. **Free tier**: the catalog must be created manually via the UI before the first deploy (see Setup step 2). +## Environments and isolation + +The bundle ships two targets, `dev` and `test`, running **the same code against +different data in different schemas**. That separation is the point, and it is +what makes end-to-end testing safe. + +| Target | Seeds from | Schemas | Purpose | +|---|---|---|---| +| `dev` | `data/` — the real seed CSVs | `dev__bronze/silver/gold` | Normal development against clean data | +| `test` | `data_test/` — corrupted fixtures | `test__bronze/silver/gold` | End-to-end validation against known defects | + +Isolation comes from one line per target: + +```yaml +presets: + name_prefix: "[test ${workspace.current_user.short_name}] " +``` + +`name_prefix` applies to **schema names as well as job names**, so it alone +separates the environments — no separate catalog, no duplicated resource +definitions, no conditional logic in the pipeline code. It is also per-user, so +two engineers on the same workspace never collide. + +### Why this matters + +A pipeline can be exercised **end to end — ingestion, transformation, +aggregation, quality assertions — with zero blast radius on real data.** A +service principal can run the whole thing on a schedule and the worst case is a +rebuilt test schema. + +That is normally the hard part of testing data pipelines. Unit tests prove the +transforms; only a full run proves Auto Loader options, checkpoint behaviour, +streaming dedup across re-runs, and the wiring between stages. Without isolation +you either don't test those, or you test them somewhere that matters. + +### Adding a preprod target + +Scaling this to a preprod environment is a target block, not an architecture +change: + +```yaml + preprod: + mode: development + presets: + name_prefix: "[preprod ${workspace.current_user.short_name}] " + tags: + environment: preprod + variables: + data_dir: data # or point at a production-volume dataset +``` + +`databricks bundle deploy --target preprod` then builds the whole medallion stack +in `preprod_*` schemas. Because it is the same code path, a preprod run measures +what production would actually do — performance at real volume, quality +expectations against real data distributions, and cost — while writing nowhere +near production tables. + +### Write-Audit-Publish + +The target pattern above is the foundation of **Write-Audit-Publish** (WAP), the +standard approach for not publishing bad data: + +| Stage | Here | +|---|---| +| **Write** | The pipeline writes to an isolated set of schemas rather than the published tables | +| **Audit** | Expectations record per-run quality metrics; `windmill_validate` asserts counts against an independently generated manifest and **fails the job** on any violation | +| **Publish** | Promote the audited output to the published location | + +This repository implements Write and Audit in full. **Publish is not +implemented** — there is no production target to promote into, so building a +promotion step would be scaffolding around a hypothetical. + +The shape it would take: run the pipeline into a staging schema, gate on the +validation job succeeding, then promote — for Delta, `CREATE OR REPLACE TABLE +prod.x DEEP CLONE staging.x`, or a catalog-level pointer swap so readers move +atomically. The audit gate is the part that already exists and is the part that +usually gets skipped. + +The two audit mechanisms are deliberately independent and neither replaces the +other. Expectations report **what the pipeline saw**; the validation job asserts +**what it should have seen**, against a manifest generated by a separate script. +An expectation cannot notice that an entire file was skipped, nor that silver +stayed invariant across repeated loads. + ## Testing The seed data is clean, so a run against it proves only the happy path. The @@ -268,7 +436,8 @@ by injecting that defect and watching it fail, rather than assuming. ## Cleanup -Remove everything the bundle created (schemas, volume, jobs) for a target: +Remove everything the bundle created (schemas, volume, jobs, pipeline) for a +target: ```bash databricks bundle destroy --target dev @@ -278,3 +447,6 @@ databricks bundle destroy --target test This deletes the tables and the dropzone volume. The catalog itself is left alone — on the free tier it was created by hand (Setup step 2), so it is not the bundle's to remove. Drop it from the UI if you want the workspace fully clean. + +Run this before checking out a different implementation branch — see +[Switching between branches](#switching-between-branches). diff --git a/transformations/01_bronze_turbine_raw.py b/transformations/01_bronze_turbine_raw.py index b38f64b..cac8bac 100644 --- a/transformations/01_bronze_turbine_raw.py +++ b/transformations/01_bronze_turbine_raw.py @@ -25,7 +25,6 @@ def turbine_raw(): .option("cloudFiles.format", "csv") .option("header", "true") .option("pathGlobFilter", "*.csv") - # Archive files once ingested so the inbox reflects what is still pending. .option("cloudFiles.cleanSource", "MOVE") .option("cloudFiles.cleanSource.moveDestination", f"{VOLUME_ROOT}/archive") .option("cloudFiles.cleanSource.retentionDuration", "7 days") From 6145ae7ecbf4cfdb9792b57a4ab69c843206bbef Mon Sep 17 00:00:00 2001 From: Pawel Bednarski Date: Mon, 20 Jul 2026 11:52:39 +0100 Subject: [PATCH 3/6] docs: simplify branch comparison and add diagrams The branch comparison read as a feature matrix aimed at someone who already knew the platform. Rewrite it around what each stage was for -- make it work, make it testable, make it declarative -- with a short "best when" table instead of eight rows of implementation detail. The trade-off that actually matters (sdp writes the least code but ties you to Databricks and needs a cutover to leave) is stated plainly rather than spread across rows. Promote the DEV and TEST targets to their own section near the top. That isolation is the point of the setup: the same code runs against different data in separate schemas, so a user or service principal can exercise the pipeline end to end without impacting real data, and preprod for full-load performance and quality testing is a config block rather than an architecture change. Add four mermaid diagrams: branch progression, environment isolation, the Write-Audit-Publish flow, and the pipeline DAG replacing the ASCII art. GitHub renders these inline. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 237 ++++++++++++++++++++++++++++++------------------------ 1 file changed, 134 insertions(+), 103 deletions(-) diff --git a/README.md b/README.md index 2e0bcdd..a4b0679 100644 --- a/README.md +++ b/README.md @@ -7,59 +7,55 @@ output has deviated from expectation. Databricks, medallion architecture (bronze → silver → gold), deployed as a Declarative Automation Bundle. This branch implements the pipeline with **Lakeflow Spark Declarative Pipelines**; two alternative implementations live on -other branches — see [Implementation branches](#implementation-branches). +other branches. ## Implementation branches -The same problem is solved three ways, so each design decision is visible as a -reviewable diff rather than asserted in prose. - -| Branch | Orchestration | Logic lives in | Data quality | -|---|---|---|---| -| [`medallion_workflow`](../../tree/medallion_workflow) | Three jobs run in sequence | Inline in notebooks | A job asserting counts after the fact | -| [`medallion_whl`](../../tree/medallion_whl) | Three jobs run in sequence | A Python wheel, unit tested | Same, plus 43 unit tests | -| **`medallion_sdp`** (this branch, and `main`) | One declarative pipeline | The same wheel | Expectations recorded per run, plus the assertion job | - -### What each stage adds - -**`medallion_workflow`** — the working pipeline. Auto Loader ingestion, streaming -silver with watermarked dedup, batch gold with anomaly detection. Introduces the -deterministic corrupted-data fixture generator and the validation job that -asserts against it. - -**`medallion_whl`** — extracts the transformation logic into `src/windmill/`, -built as a wheel by the bundle and attached to jobs. Notebooks become -orchestration only. The logic gains unit tests that run against small local -DataFrames in ~11 seconds with no cluster — a loop fast enough to actually catch -statistical bugs, which the integration harness is too slow to do. - -**`medallion_sdp`** — replaces the three jobs with one pipeline. Execution order -is derived from table references rather than declared; the runtime owns -checkpoints, triggers, output modes and writes. Row-level checks become -expectations, so quality is a queryable per-run artifact instead of a bespoke job. - -### Comparison - -| | `medallion_workflow` | `medallion_whl` | `medallion_sdp` | -|---|---|---|---| -| Lines of orchestration | Most | Same | Least | -| Execution order | Declared by running jobs in sequence | Same | Derived from table references | -| Checkpoints, triggers, writes | Hand-managed per stream | Same | Owned by the runtime | -| Unit testable without a cluster | No | **Yes** | **Yes** | -| Quality metrics per run | No | No | **Yes**, in the event log | -| Table ownership | Plain managed Delta | Same | Pipeline-owned | -| Migration in / out | Free | Free | **Cutover required** | -| Portability | Plain PySpark + any scheduler | Same | Databricks-specific | - -None is strictly best. `medallion_whl` is the strongest general engineering -artifact — packaged, tested, portable. `medallion_sdp` is the better choice when -the pipeline shape is stable and quality reporting matters, at the cost of being -tied to the platform and harder to migrate away from. `medallion_workflow` is the -easiest to reason about while the logic is still moving. - -All three produce **identical results** on the same input: silver 11,169 rows, 54 -invalid readings, 465 turbine-days, 20 anomalies. The stage 1 validation job -passes unchanged on all three, which is what makes the comparison fair. +The same problem is solved three ways. Each branch is a working solution — they +build on each other, so the design decisions are visible as diffs. + +```mermaid +flowchart LR + A["medallion_workflow
Make it work"] + B["medallion_whl
Make it testable"] + C["medallion_sdp
Make it declarative"] + A --> B --> C + C -.-> M["main"] +``` + +**1. `medallion_workflow` — make it work.** +Three notebooks, run one after another as jobs. Straightforward to read top to +bottom. The logic lives inside the notebooks, so testing it means running the +whole thing on a cluster. + +**2. `medallion_whl` — make it testable.** +The same pipeline, but the logic moves out of the notebooks into a small Python +package, packaged as a wheel. The notebooks shrink to "read, transform, write". +Now the logic can be tested in **11 seconds on a laptop** instead of minutes on a +cluster — fast enough to actually catch bugs while writing them. + +**3. `medallion_sdp` — make it declarative.** +The three jobs collapse into one pipeline. Instead of telling Databricks *when* +to run each step, you declare the tables and it works out the order. Data quality +rules become **expectations**, so every run automatically records how many rows +passed and failed. + +### Which would I use? + +| | Best when | +|---|---| +| `medallion_workflow` | The logic is still changing and you want the simplest thing to read | +| `medallion_whl` | You want fast tests and code that runs anywhere, not just Databricks | +| `medallion_sdp` | The shape is settled and you want quality reporting for free | + +The honest trade: **`medallion_sdp` writes the least code but ties you to +Databricks**, and its tables belong to the pipeline, so moving away later means a +planned cutover rather than a switch. `medallion_whl` is plain PySpark — it would +run anywhere with a scheduler. + +**All three produce identical results** — 11,169 clean rows, 54 invalid readings, +465 turbine-days, 20 anomalies — and the same validation job passes on all three. +That is what makes comparing them meaningful rather than three separate claims. ### Switching between branches @@ -138,32 +134,42 @@ databricks bundle run --target dev windmill_pipeline ## Architecture ### Data Flow + +```mermaid +flowchart TD + CSV["data_group_*.csv"] --> INIT["windmill_init
copies files into the volume"] + INIT --> INBOX[("dropzone/inbox")] + + subgraph PIPE["windmill_pipeline"] + BRONZE["turbine_raw
streaming table
Auto Loader, raw as delivered"] + SILVER["turbine_clean
streaming table
dedup + validity expectations"] + GOLD["turbine_summary
materialized view
daily stats + anomalies"] + BRONZE --> SILVER --> GOLD + end + + INBOX --> BRONZE + GOLD --> VALIDATE["windmill_validate
asserts against the manifest"] ``` -data_group_*.csv (Raw) - ↓ -[windmill_init] → Volume: dropzone/inbox - ↓ -┌─ windmill_pipeline ─────────────────────────────────┐ -│ turbine_raw (streaming table, Auto Loader) │ -│ ↓ │ -│ turbine_clean (streaming table, + expectations) │ -│ ↓ │ -│ turbine_summary (materialized view, aggregate) │ -└─────────────────────────────────────────────────────┘ - -Order is derived from the table references in the code, not declared. -``` + +The order inside the pipeline is **derived from the table references in the +code**, not declared anywhere. `windmill_init` and `windmill_validate` stay as +jobs because getting files into a volume, and asserting on the result, are not +the pipeline's concern. ### Catalog Structure + +Each target gets its own copy of this, prefixed — `dev_user_bronze`, +`test_user_bronze`, and so on. + ``` windmill_classcell/ -├── bronze/ -│ ├── dropzone (volume) -│ └── turbine_raw (table) -├── silver/ -│ └── turbine_clean (table) -└── gold/ - └── turbine_summary (table) +├── _bronze/ +│ ├── dropzone (volume) ← inbox, archive, checkpoints +│ └── turbine_raw +├── _silver/ +│ └── turbine_clean +└── _gold/ + └── turbine_summary ``` ## Data Processing @@ -301,45 +307,63 @@ windmill_classcell/ 7. **Free tier**: the catalog must be created manually via the UI before the first deploy (see Setup step 2). -## Environments and isolation +## Environments: DEV and TEST -The bundle ships two targets, `dev` and `test`, running **the same code against -different data in different schemas**. That separation is the point, and it is -what makes end-to-end testing safe. +**This is the part that matters most.** -| Target | Seeds from | Schemas | Purpose | -|---|---|---|---| -| `dev` | `data/` — the real seed CSVs | `dev__bronze/silver/gold` | Normal development against clean data | -| `test` | `data_test/` — corrupted fixtures | `test__bronze/silver/gold` | End-to-end validation against known defects | +The bundle ships two targets — `dev` and `test` — running the **same code against +different data in completely separate schemas**. A user or a service principal +can run the pipeline end to end **without impacting any real data.** -Isolation comes from one line per target: +```mermaid +flowchart TD + CODE["One codebase
one bundle"] + + CODE --> DEV["dev
clean seed data"] + CODE --> TEST["test
deliberately broken data"] + CODE -.-> PRE["preprod
production-volume data
(add when needed)"] + + DEV --> DEVS[("dev_user_bronze
dev_user_silver
dev_user_gold")] + TEST --> TESTS[("test_user_bronze
test_user_silver
test_user_gold")] + PRE -.-> PRES[("preprod_user_bronze
preprod_user_silver
preprod_user_gold")] + + DEVS --> SAFE["No shared tables.
Nothing to break."] + TESTS --> SAFE + PRES -.-> SAFE +``` + +| Target | Data it uses | What it is for | +|---|---|---| +| **dev** | The real CSVs, clean | Everyday development | +| **test** | Corrupted fixtures with known defects | Proving the pipeline catches bad data | + +### How the separation works + +One line per target does it: ```yaml presets: name_prefix: "[test ${workspace.current_user.short_name}] " ``` -`name_prefix` applies to **schema names as well as job names**, so it alone -separates the environments — no separate catalog, no duplicated resource -definitions, no conditional logic in the pipeline code. It is also per-user, so -two engineers on the same workspace never collide. +That prefix lands on **schema names as well as job names**. No second catalog, no +duplicated resource files, no `if environment == "test"` anywhere in the pipeline +code. It also includes the username, so two engineers on the same workspace never +collide. -### Why this matters +### Why this is worth doing -A pipeline can be exercised **end to end — ingestion, transformation, -aggregation, quality assertions — with zero blast radius on real data.** A -service principal can run the whole thing on a schedule and the worst case is a -rebuilt test schema. +Unit tests prove the transformation logic. They cannot prove file ingestion, +checkpoint behaviour, deduplication across repeated loads, or the wiring between +stages — **only a real end-to-end run proves those.** -That is normally the hard part of testing data pipelines. Unit tests prove the -transforms; only a full run proves Auto Loader options, checkpoint behaviour, -streaming dedup across re-runs, and the wiring between stages. Without isolation -you either don't test those, or you test them somewhere that matters. +Without isolated targets you have two bad options: skip that testing, or do it +somewhere that matters. With them, the worst possible outcome of a broken run is +a rebuilt test schema. -### Adding a preprod target +### Adding preprod for full-load testing -Scaling this to a preprod environment is a target block, not an architecture -change: +Scaling this to preprod is a config block, not an architecture change: ```yaml preprod: @@ -352,16 +376,23 @@ change: data_dir: data # or point at a production-volume dataset ``` -`databricks bundle deploy --target preprod` then builds the whole medallion stack -in `preprod_*` schemas. Because it is the same code path, a preprod run measures -what production would actually do — performance at real volume, quality -expectations against real data distributions, and cost — while writing nowhere -near production tables. +`databricks bundle deploy --target preprod` builds the entire stack in `preprod_*` +schemas. Because it is the identical code path, a preprod run tells you what +production would actually do — **performance at real volume, data quality against +real distributions, and cost** — while writing nowhere near production tables. ### Write-Audit-Publish -The target pattern above is the foundation of **Write-Audit-Publish** (WAP), the -standard approach for not publishing bad data: +Isolated targets are the foundation of **Write-Audit-Publish** (WAP) — the +standard approach to not publishing bad data. + +```mermaid +flowchart LR + SRC["Incoming CSVs"] --> W["WRITE
pipeline runs into
an isolated schema"] + W --> A{"AUDIT
expectations +
assertion job"} + A -->|"passes"| P["PUBLISH
promote to
production tables"] + A -->|"fails"| STOP["Stop.
Bad data never
reaches consumers."] +``` | Stage | Here | |---|---| From 677fab1ed752ce7655a3fe14886d48e483e409ec Mon Sep 17 00:00:00 2001 From: Pawel Bednarski Date: Mon, 20 Jul 2026 12:23:32 +0100 Subject: [PATCH 4/6] docs: add TL;DR and a session transcript exporter The task asks for a brief description of the solution design and the assumptions made. The README opened straight into branch detail, so a reader had to work through several sections before reaching either. Add a TL;DR covering what was built, why it exists in three implementations, the environment separation that makes end-to-end testing safe, and the two decisions that shape everything else -- flagging invalid readings rather than dropping or imputing them, and computing anomaly baselines over daily means rather than raw readings. scripts/export_chat.py converts a Claude Code session transcript to Markdown. The raw JSONL is ~4.5 MB of interleaved prose, tool calls and tool output; verbatim it is unreadable, so prose is kept in full while tool input and output are truncated to a collapsed preview. Harness-injected blocks are stripped, and sub-agent threads are skipped. The generated docs/opus_chat.md is gitignored. It is a working record rather than part of the deliverable, and it carries the full back-and-forth instead of the conclusions -- which are what the docs are for. Regenerate it locally when wanted. Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 5 ++ README.md | 35 ++++++-- scripts/export_chat.py | 191 +++++++++++++++++++++++++++++++++++++++++ 3 files changed, 226 insertions(+), 5 deletions(-) create mode 100644 scripts/export_chat.py diff --git a/.gitignore b/.gitignore index 269cac2..e6b7785 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,8 @@ build/ .env.local dist/ .venv/ + +# Session transcript — regenerate with scripts/export_chat.py. +# Kept local: it is a working record, not part of the deliverable, and it +# contains the full back-and-forth rather than the conclusions. +docs/opus_chat.md diff --git a/README.md b/README.md index a4b0679..f641309 100644 --- a/README.md +++ b/README.md @@ -4,10 +4,34 @@ Ingests raw turbine measurements delivered as daily CSVs, cleans and validates them, computes per-turbine summary statistics, and identifies turbines whose output has deviated from expectation. -Databricks, medallion architecture (bronze → silver → gold), deployed as a -Declarative Automation Bundle. This branch implements the pipeline with -**Lakeflow Spark Declarative Pipelines**; two alternative implementations live on -other branches. +## TL;DR + +A PySpark pipeline on Databricks, medallion architecture (bronze → silver → +gold), deployed as a Declarative Automation Bundle. It exists in **three +implementations** on three branches — jobs with notebooks, the same jobs with the +logic extracted into a tested Python wheel, and a declarative pipeline — because +the interesting part of this problem is the design trade-offs, and the clearest +way to show a trade-off is to build both sides. All three produce identical +results and pass the same validation. + +It deploys into **separate environments** (`dev`, `test`, and a `preprod` you can +add in ten lines) that run **the same code against different data in isolated +schemas**. That separation is the backbone: a person or a service principal can +exercise the pipeline end to end — ingestion, transformation, quality assertions +— **without touching anything real**, which is what makes testing a data pipeline +possible at all. + +Two decisions shape the rest. **Invalid readings are flagged, never dropped or +imputed** — a forward-filled sensor gap is an invented measurement, and a deleted +outlier is the deliverable thrown away. And **anomaly baselines are computed over +daily means, not raw readings**, because averaging 24 hourly values shrinks the +spread by √24; getting that wrong made the first version mathematically incapable +of reporting a single anomaly. Full reasoning in +[Assumptions](#assumptions) and [`docs/`](docs/). + +> This branch (`main`) implements the pipeline with **Lakeflow Spark Declarative +> Pipelines**. See [Implementation branches](#implementation-branches) for the +> other two. ## Implementation branches @@ -249,7 +273,8 @@ windmill_classcell/ │ ├── 01_copy_data.py # Seeds the dropzone volume │ └── 05_validate.py # Asserts pipeline output against the manifest ├── scripts/ -│ └── generate_test_data.py +│ ├── generate_test_data.py +│ └── export_chat.py # Session transcript → docs/opus_chat.md (gitignored) └── docs/ ├── architecture.md ├── declarative_pipeline.md diff --git a/scripts/export_chat.py b/scripts/export_chat.py new file mode 100644 index 0000000..b49bf61 --- /dev/null +++ b/scripts/export_chat.py @@ -0,0 +1,191 @@ +#!/usr/bin/env python3 +"""Convert a Claude Code session transcript (JSONL) into readable Markdown. + +The raw transcript is ~4.5 MB of interleaved messages, tool calls and tool +results. Dumping it verbatim would be unreadable, so tool inputs and outputs are +truncated to a preview while user and assistant prose is kept in full -- the +reasoning is the interesting part, not the thousandth line of a table dump. + +Usage: + python3 scripts/export_chat.py [transcript.jsonl] [-o docs/opus_chat.md] + +With no arguments it picks the most recently modified transcript for this +project from ~/.claude/projects/. +""" + +from __future__ import annotations + +import argparse +import json +import pathlib +import sys +from datetime import datetime + +PROJECTS = pathlib.Path.home() / ".claude" / "projects" +MAX_TOOL_INPUT = 900 +MAX_TOOL_RESULT = 700 +# Injected by the harness rather than typed by anyone; noise in a transcript. +NOISE_MARKERS = ( + "", + "CAVEMAN MODE ACTIVE", + "UserPromptSubmit hook", + "SessionStart hook", + "[DATABRICKS]", +) + + +def find_latest_transcript() -> pathlib.Path: + candidates = list(PROJECTS.glob("*/*.jsonl")) + if not candidates: + raise SystemExit(f"no transcripts found under {PROJECTS}") + return max(candidates, key=lambda p: p.stat().st_mtime) + + +def clip(text: str, limit: int) -> str: + text = text.rstrip() + if len(text) <= limit: + return text + return text[:limit].rstrip() + f"\n… [{len(text) - limit:,} more characters]" + + +def is_noise(text: str) -> bool: + stripped = text.strip() + if not stripped: + return True + return any(m in stripped for m in NOISE_MARKERS) + + +def strip_noise(text: str) -> str: + """Drop harness-injected blocks while keeping what the user actually wrote.""" + out = [] + for para in text.split("\n\n"): + if not is_noise(para): + out.append(para) + return "\n\n".join(out).strip() + + +def render(path: pathlib.Path) -> str: + lines: list[str] = [] + first_ts = last_ts = None + counts = {"user": 0, "assistant": 0, "tools": 0} + + for raw in path.open(): + raw = raw.strip() + if not raw: + continue + try: + rec = json.loads(raw) + except json.JSONDecodeError: + continue + + if rec.get("type") not in ("user", "assistant"): + continue + # Sub-agent chatter belongs to its own thread, not the main transcript. + if rec.get("isSidechain"): + continue + + ts = rec.get("timestamp") + if ts: + first_ts = first_ts or ts + last_ts = ts + + msg = rec.get("message") or {} + role = msg.get("role") + content = msg.get("content") + if isinstance(content, str): + content = [{"type": "text", "text": content}] + if not isinstance(content, list): + continue + + for block in content: + if not isinstance(block, dict): + continue + btype = block.get("type") + + if btype == "text": + text = strip_noise(block.get("text", "")) + if not text: + continue + if role == "user": + counts["user"] += 1 + lines.append(f"\n---\n\n### 👤 User\n\n{text}\n") + else: + counts["assistant"] += 1 + lines.append(f"\n### 🤖 Assistant\n\n{text}\n") + + elif btype == "tool_use": + counts["tools"] += 1 + name = block.get("name", "?") + params = block.get("input", {}) or {} + # Show the parameter that says what the call actually did. + key = next( + (k for k in ("command", "file_path", "url", "prompt", "skill", "query") + if k in params), + None, + ) + detail = str(params.get(key, "")) if key else json.dumps(params)[:MAX_TOOL_INPUT] + lines.append( + f"
\n🔧 {name}\n\n" + f"```\n{clip(detail, MAX_TOOL_INPUT)}\n```\n\n
\n" + ) + + elif btype == "tool_result": + res = block.get("content") + if isinstance(res, list): + res = "\n".join( + b.get("text", "") for b in res + if isinstance(b, dict) and b.get("type") == "text" + ) + res = (res or "").strip() if isinstance(res, str) else "" + if not res: + continue + lines.append( + f"
\n📄 Result\n\n" + f"```\n{clip(res, MAX_TOOL_RESULT)}\n```\n\n
\n" + ) + + def fmt(t): + if not t: + return "?" + try: + return datetime.fromisoformat(t.replace("Z", "+00:00")).strftime("%Y-%m-%d %H:%M UTC") + except ValueError: + return t + + header = [ + "# Session transcript", + "", + "Working session that produced this repository: the medallion pipeline, the", + "wheel refactor with unit tests, and the declarative rewrite.", + "", + "Exported with `scripts/export_chat.py`. Prose is verbatim; tool calls and", + "their output are truncated to a preview and collapsed, since the reasoning", + "matters more than the command output.", + "", + f"- **Source**: `{path.name}`", + f"- **Span**: {fmt(first_ts)} → {fmt(last_ts)}", + f"- **Messages**: {counts['user']} user, {counts['assistant']} assistant", + f"- **Tool calls**: {counts['tools']}", + "", + ] + return "\n".join(header) + "\n".join(lines) + "\n" + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("transcript", nargs="?", type=pathlib.Path) + ap.add_argument("-o", "--output", type=pathlib.Path, default=pathlib.Path("docs/opus_chat.md")) + args = ap.parse_args() + + src = args.transcript or find_latest_transcript() + if not src.exists(): + raise SystemExit(f"not found: {src}") + + args.output.parent.mkdir(parents=True, exist_ok=True) + md = render(src) + args.output.write_text(md) + print(f"{src}\n -> {args.output} ({len(md):,} chars, {md.count(chr(10)):,} lines)") + + +if __name__ == "__main__": + main() From 016b8b9fe92d09312ca474532b1232dfcaec0008 Mon Sep 17 00:00:00 2001 From: Pawel Bednarski Date: Mon, 20 Jul 2026 12:29:44 +0100 Subject: [PATCH 5/6] update authors README --- README.md | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f641309..6a2a7b2 100644 --- a/README.md +++ b/README.md @@ -4,6 +4,10 @@ Ingests raw turbine measurements delivered as daily CSVs, cleans and validates them, computes per-turbine summary statistics, and identifies turbines whose output has deviated from expectation. +author: p17b +llm ="claude opus 4.8 xhigh" +ide = pycharm, cmux + ## TL;DR A PySpark pipeline on Databricks, medallion architecture (bronze → silver → @@ -14,8 +18,8 @@ the interesting part of this problem is the design trade-offs, and the clearest way to show a trade-off is to build both sides. All three produce identical results and pass the same validation. -It deploys into **separate environments** (`dev`, `test`, and a `preprod` you can -add in ten lines) that run **the same code against different data in isolated +It deploys into **separate environments** (`dev`, `test`) +that run **the same code against different data in isolated schemas**. That separation is the backbone: a person or a service principal can exercise the pipeline end to end — ingestion, transformation, quality assertions — **without touching anything real**, which is what makes testing a data pipeline From 040caed6c46ccb5847e4073da36afb372f5e9c01 Mon Sep 17 00:00:00 2001 From: Pawel Bednarski Date: Mon, 20 Jul 2026 13:00:15 +0100 Subject: [PATCH 6/6] docs: rewrite architecture doc to match the implementation docs/architecture.md was written during initial scaffolding and never updated. It asserted behaviour the pipeline does not have and, in two places, the exact opposite of the documented design: Silver Layer: Forward-fill missing values (by turbine) Silver Layer: Outlier removal (>2 std dev from mean) Both were deliberately rejected -- forward-filling invents measurements that were never taken, and removing outliers deletes the deliverable the task asks for. It also listed an efficiency score that was never implemented, and an assumption that sensor gaps are forward-fillable. A reader would have concluded the pipeline does the reverse of what the README argues for. Same class of problem as the dead src/ package removed earlier: documentation drifting into a claim about behaviour. Rewritten around the decisions and what was rejected, since the README already covers what it does and how to run it. Adds the silver/gold split rationale (a check belongs in silver if one row can judge it), the sqrt(24) baseline defect and its fix, per-turbine rather than fleet-wide baselines, the >= 0 boundary, why gold is a materialized view rather than a streaming table, and an honest known- gaps section. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/architecture.md | 191 +++++++++++++++++++++++++++---------------- 1 file changed, 120 insertions(+), 71 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index 2206c37..08b006f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1,97 +1,146 @@ -# Windmill Pipeline Architecture +# Architecture -## Overview +Why the pipeline is shaped the way it is. The README covers what it does and how +to run it; this covers the decisions and what was rejected. -Windmill is a Databricks demo project implementing a medallion architecture (bronze/silver/gold) for renewable energy turbine data processing. +## Medallion layers -## Design Decisions +| Layer | Holds | Rule | +|---|---|---| +| Bronze | Raw readings exactly as delivered, plus ingestion metadata | Never filtered, never corrected | +| Silver | Deduplicated readings with per-field validity flags | Row-level judgements only | +| Gold | Daily per-turbine statistics and anomaly flags | Anything needing a population | -### 1. Medallion Architecture -- **Bronze**: Raw data ingestion with minimal transformation -- **Silver**: Cleaned, validated, and deduplicated data -- **Gold**: Business-ready aggregations and analytics +The split that matters is between **silver and gold**, and it is not about +cleanliness — it is about how much context a check needs. -**Why**: Industry standard for data lakes. Clear separation of concerns, easy to understand and audit. +A check belongs in **silver** if a single row is enough to judge it: is +`power_output` null, is it negative, is this turbine reporting from a file it +does not belong to. No other row is required to answer. -### 2. Databricks Asset Bundles (DABs) -- Infrastructure as Code for entire pipeline -- Single deployment command -- Idempotent, version-controlled +A check belongs in **gold** if it needs a population: is this turbine-day +unusual *relative to that turbine's own history*. That cannot be answered one row +at a time, and silver's watermark exists to bound deduplication state, not to +define a statistical window. -**Why**: Demo-friendly. Recruiter can deploy in one command. Direct engine (no Terraform dependency). +## Decisions -### 3. Serverless SQL -- No compute management -- Automatic scaling -- Cost-effective for demos +### Invalid readings are flagged, not dropped or imputed -**Why**: Free tier accessible. No cluster configuration needed. +Silver marks each field with a boolean and carries the row through unchanged. +Gold filters on `is_reading_valid` before computing statistics. -### 4. Data Volume for Landing Zone -- CSV files uploaded to volume -- Acts as staging area -- Immutable source of truth +Rejected: **forward-filling gaps**, which invents measurements that were never +taken and makes a broken sensor look like a working one. Rejected: **dropping +invalid rows**, which destroys the evidence someone needs to diagnose the sensor +and makes "how much data did we lose?" unanswerable. -**Why**: Separates raw data from processed tables. Easy to re-run pipeline. +Quarantining in place gives gold clean statistics and keeps quality reporting +possible. On the declarative branch the same rule appears as `expect_all` +expectations — warn and keep, never `_or_drop` or `_or_fail`. -## Data Flow +### Anomalies are flagged, not removed -``` -Raw CSV → Volume (dropzone) → Bronze Table → Silver Table → Gold Table - ↓ ↓ ↓ ↓ - [01_copy_data] [02_bronze] [03_silver] [04_gold] -``` +The task asks for turbines that have deviated to be *identified*. Filtering them +out upstream would delete the deliverable. Gold emits `is_anomaly` alongside the +bounds that produced it, so a reviewer can see why a day tripped. -## Quality Checks +### Baselines are computed over daily means, not raw readings -### Bronze Layer -- Schema validation -- Deduplication (timestamp + turbine_id) -- Metadata tracking (ingestion_timestamp, source_file) +The most consequential decision here, and it was originally wrong. -### Silver Layer -- Forward-fill missing values (by turbine) -- Outlier removal (>2 std dev from mean) -- Metadata preservation +Averaging 24 hourly readings shrinks the spread by √24 ≈ 5 — the standard error +of the mean. A band built from reading-level standard deviation is therefore +about five times too wide for a daily mean to ever escape. Measured on the seed +data: reading-level σ ≈ 0.86 against daily-mean σ ≈ 0.17, giving a band of +[1.30, 4.74] versus an actual daily range of [2.47, 3.38]. -### Gold Layer -- 24-hour summary statistics (min, max, avg, stddev) -- Power output anomaly detection -- Efficiency score calculation +The first implementation compared daily averages against reading-level spread and +reported **zero anomalies across 465 turbine-days** — not because the data was +clean, but because the test was mathematically incapable of firing. Full +diagnosis in [`dev_seed_original_results.md`](dev_seed_original_results.md). -## Scalability +The rule: the baseline must be built from the same unit being tested against it. -### Current (Demo) -- 15 turbines, hourly data -- CSV file-based ingestion -- 3 separate jobs (can be orchestrated) +### Baselines are per-turbine, not fleet-wide -### Production Evolution -1. Add job orchestration (Databricks Jobs) -2. Delta Live Tables (DLT) for pipeline management -3. Streaming ingestion (Kafka/Kinesis) -4. Time-series analytics (materialized views) -5. ML models (anomaly detection, forecasting) +Turbines sit in different wind conditions. A fleet baseline conflates "this +turbine is underperforming" with "this turbine is in a calmer spot", and would +report a well-sited turbine as permanently anomalous. The task asks for deviation +from *their* expected output. -## Assumptions +### Validity bounds are `>= 0`, not `> 0` -1. **Data**: Hourly readings per turbine -2. **Files**: 5 turbines per CSV, consistent schema -3. **Missing Data**: Forward-fillable sensor gaps -4. **Outliers**: >2 σ from mean = anomaly -5. **Auth**: OAuth (OAuth preferred over PAT) -6. **Cost**: FREE tier Databricks workspace +A becalmed or idling turbine genuinely reports 0 MW in 0 m/s wind. Treating zero +as invalid would discard real low-wind data silently. Guarded by unit tests that +fail if the bound is ever tightened. -## Testing +## Dataset types (declarative branch) + +Bronze and silver are **streaming tables**: append-only, processing only what +arrived since the last run. + +Gold is a **materialized view**. Streaming tables never revisit rows they have +already emitted, so a daily aggregate built as one would go stale the moment late +or corrected data reached silver. A materialized view recomputes. It reads silver +with `spark.read`, not `readStream` — a streaming read would impose watermark and +state constraints for no benefit, since the anomaly baseline spans the whole +history rather than a window. -- Unit tests for transformation logic (pytest) -- Integration tests via Databricks notebooks -- Data quality assertions in each layer +## Environment separation -## Future Enhancements +Targets (`dev`, `test`, and any others added) run the same code against different +data in separate schemas, isolated by a single `name_prefix` preset that applies +to schema names as well as job names. + +This is what makes end-to-end testing possible. Unit tests prove the transformation +logic; only a real run proves Auto Loader behaviour, checkpoints, deduplication +across repeated loads, and the wiring between stages. Without isolation those go +untested or get tested somewhere that matters. + +## Testing -1. **Orchestration**: DABs job scheduling -2. **Monitoring**: Databricks Jobs SQL alerts -3. **ML**: MLflow for anomaly detection models -4. **Real-time**: Structured Streaming for live data -5. **API**: Databricks Apps for stakeholder dashboards +Two layers, covering different failures, neither replacing the other. + +**Unit tests** (43, ~11s, no cluster) cover statistical behaviour and edge cases +that are awkward to express as fixtures — null spread, single-day turbines, +boundary values, per-turbine versus fleet-wide baselines. Possible because the +transformation logic is packaged separately from the notebooks as pure +DataFrame-in / DataFrame-out functions. + +**Integration** runs the whole pipeline against deliberately corrupted fixtures +with known expected counts, generated deterministically and asserted against a +manifest. Covers what unit tests cannot see: file ingestion, checkpoints, +load-invariance, job wiring. + +See [`testing.md`](testing.md), including the mutation check that caught a +regression guard which passed while the defect was present. + +## Scale + +Currently 15 turbines at hourly grain — 11,160 rows for a month. Nothing here is +sized for that: Auto Loader ingests incrementally by file, silver processes only +new bronze rows, and the volume of a real farm changes the runtime rather than +the design. + +The parts that would need attention first at genuinely large scale: + +- **Gold recomputes the full baseline each run.** Fine at 465 turbine-days; + at millions it would want incremental refresh or a rolling window. +- **Deduplication state** is bounded by a 2-day watermark. Higher throughput + means more state per micro-batch, and the watermark becomes a real tuning knob + rather than a formality. +- **Liquid clustering** on `(turbine_id, date)` in gold, once table size makes + file pruning matter. + +## Known gaps + +- **Publish step of Write-Audit-Publish is not implemented** — there is no + production target to promote into. +- **Schema drift is untested.** Auto Loader runs on a fixed schema; a file with a + renamed or extra column has an unobserved failure mode. +- **Late-arriving data beyond the watermark** is dropped by design, but there is + no explicit test recording that as intentional. +- **Sub-hourly readings** are detected (`has_sub_hourly_readings`) but not + corrected — the right correction depends on sensor semantics that cannot be + determined from the data. See the assumptions section of the README.