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.

author: p17b llm ="claude opus 4.8 xhigh" ide = pycharm, cmux

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)
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 and docs/.

This branch (main) implements the pipeline with Lakeflow Spark Declarative Pipelines. See Implementation branches for the other two.

Implementation branches

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.

flowchart LR
    A["medallion_workflow<br/><i>Make it work</i>"]
    B["medallion_whl<br/><i>Make it testable</i>"]
    C["medallion_sdp<br/><i>Make it declarative</i>"]
    A --> B --> C
    C -.-> M["main"]

Loading

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

Branches are not drop-in replacements for each other in a workspace that already has a deployment. Destroy first:

databricks bundle destroy --target dev      # and --target test
git checkout <branch>
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

Prerequisites

• Databricks account (FREE tier or commercial)
• Databricks CLI v0.279.0+
• Python 3.10+
• Unity Catalog enabled with default metastore configured OR catalog windmill_classcell created manually via UI

Setup

1. Clone repo

cd ~/code/lab/windmill_classcell

2. Create catalog (FREE tier only) If using FREE edition or default metastore not enabled:

• Go to Databricks UI → Data → Catalogs → Create
• Name: windmill_classcell
• Select "Use Default Storage"
• Create

If default metastore enabled, skip this step.

3. Authenticate with Databricks

databricks auth login --host https://<your-workspace>.cloud.databricks.com

databricks.yml deliberately does not pin a workspace host — the CLI resolves it from the profile this command writes to ~/.databrickscfg, so the repo runs against your workspace with no edits. To target a specific profile, add --profile <name> to the commands below, or export DATABRICKS_HOST.

4. Deploy bundle (creates schemas, volumes, jobs)

databricks bundle deploy --target dev

5. Run initialization job (copies CSV data to volume)

databricks bundle run --target dev windmill_init

6. Run the pipeline (bronze, silver and gold in one update)

databricks bundle run --target dev windmill_pipeline

Architecture

Data Flow

flowchart TD
    CSV["data_group_*.csv"] --> INIT["windmill_init<br/><i>copies files into the volume</i>"]
    INIT --> INBOX[("dropzone/inbox")]

    subgraph PIPE["windmill_pipeline"]
        BRONZE["<b>turbine_raw</b><br/>streaming table<br/><i>Auto Loader, raw as delivered</i>"]
        SILVER["<b>turbine_clean</b><br/>streaming table<br/><i>dedup + validity expectations</i>"]
        GOLD["<b>turbine_summary</b><br/>materialized view<br/><i>daily stats + anomalies</i>"]
        BRONZE --> SILVER --> GOLD
    end

    INBOX --> BRONZE
    GOLD --> VALIDATE["windmill_validate<br/><i>asserts against the manifest</i>"]

Loading

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/
├── <target>_bronze/
│   ├── dropzone (volume)   ← inbox, archive, checkpoints
│   └── turbine_raw
├── <target>_silver/
│   └── turbine_clean
└── <target>_gold/
    └── turbine_summary

Data Processing

Bronze Layer

Raw landing, no interpretation. Auto Loader (cloudFiles) streams CSVs from the dropzone volume with an explicit schema — no inference, so a malformed file cannot silently change column types. Adds ingestion metadata (source_file, timestamps). Files are archived after ingestion via cloudFiles.cleanSource. Append-only: nothing is filtered or corrected here.

Silver Layer

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
• turbine_group extracted from the source filename, UNKNOWN when it does not match data_group_<digit>
• is_turbine_id_valid — turbine falls in its group's expected range
• is_power_output_valid, is_wind_speed_valid, is_wind_direction_valid — null and range checks, bounds >= 0
• is_reading_valid — rollup of the above, used by gold to exclude bad rows from statistics

Invalid rows are flagged in place, not dropped or imputed. Statistical 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

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 matches the unit the threshold was derived from
• is_anomaly — daily mean outside 2σ of that turbine's own baseline
• anomalous_reading_count — individual readings breaching the reading-level band, catching short spikes that average out over a day

Project Structure

windmill_classcell/
├── databricks.yml          # Bundle config: variables, artifacts, includes, targets
├── pyproject.toml          # Wheel build + dev dependencies
├── src/windmill/           # Transformation logic, shipped as a wheel
│   ├── schema.py           # Schema and domain constants
│   ├── validation.py       # Silver: dedup and row-level validity
│   └── aggregation.py      # Gold: daily statistics and anomaly detection
├── tests/                  # Unit tests, local Spark, no cluster needed
│   ├── conftest.py
│   ├── test_validation.py
│   └── test_aggregation.py
├── resources/              # One file per resource, <name>.<resource_type>.yml
│   ├── bronze.schema.yml
│   ├── silver.schema.yml
│   ├── gold.schema.yml
│   ├── dropzone.volume.yml
│   ├── windmill.pipeline.yml
│   ├── windmill_init.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
│   ├── data_group_1.csv
│   ├── data_group_2.csv
│   └── data_group_3.csv
├── data_test/              # Corrupted fixtures + expectation manifest (generated)
├── 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
│   └── export_chat.py      # Session transcript → docs/opus_chat.md (gitignored)
└── docs/
    ├── architecture.md
    ├── declarative_pipeline.md
    ├── testing.md
    └── dev_seed_original_results.md

Assumptions

1. The feed delivers exactly one reading per turbine per hour, on the hour. True of the provided data — all 11,160 timestamps land on :00:00, there are no duplicate (timestamp, turbine_id) pairs, and every one of the 465 turbine-days has exactly 24 readings.

   This is load-bearing in two places. Deduplication treats (timestamp, turbine_id) as the natural key, so it collapses re-delivered rows but correctly leaves genuinely distinct sub-hour readings alone. And avg_power_output is an unweighted mean over readings, which is only right while each hour contributes one sample.

   If the feed ever sent finer-grained timestamps — say two readings at 00:15 and 00:47 — nothing would break, but the daily mean would quietly reweight toward the busier hours. Gold therefore emits distinct_hours alongside measurement_count and sets has_sub_hourly_readings when they disagree, so the condition is visible instead of silent.

   The right correction depends on what the sensor actually emits, which cannot be determined from the data:

   If each reading is…	Correct treatment
   An instantaneous sample	Average them (what the pipeline does today)
   A cumulative counter	Take the last value in the hour, and difference it
   A pre-averaged period value	Weight by the period each covers

   The spec states output is measured in megawatts — a rate, not an energy total — which points at the first reading and makes averaging defensible. But that is inference, not confirmation. Rather than guess at a resampling rule, the pipeline holds the assumption explicitly and flags violations; the rule can be implemented once someone who knows the sensor confirms the semantics.

2. 5 turbines per CSV file (3 files = 15 turbines), and a turbine always appears in the same file — so the filename is a usable source of group identity.

3. Invalid readings are flagged, not dropped or imputed. Nulls and negatives are marked with per-field boolean columns and carried through to silver. Forward-filling a sensor gap would invent measurements that were never taken; quarantining in place keeps the rows available for quality reporting and sensor root-cause work, and lets gold exclude them from statistics.

4. Anomalies are flagged, not removed. The spec asks for deviating turbines to be identified, so filtering them out in silver would destroy the deliverable. Threshold is 2 standard deviations from that turbine's own baseline, computed over daily means — see docs/architecture.md for why the unit matters.

5. A reading of 0 is valid: a becalmed or idling turbine genuinely reports 0 MW. Validity bounds are >= 0, not > 0.

6. Serverless compute; the workspace used for development is serverless-only.

7. Free tier: the catalog must be created manually via the UI before the first deploy (see Setup step 2).

Environments: DEV and TEST

This is the part that matters most.

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.

flowchart TD
    CODE["One codebase<br/>one bundle"]

    CODE --> DEV["<b>dev</b><br/>clean seed data"]
    CODE --> TEST["<b>test</b><br/>deliberately broken data"]
    CODE -.-> PRE["<b>preprod</b><br/>production-volume data<br/><i>(add when needed)</i>"]

    DEV --> DEVS[("dev_user_bronze<br/>dev_user_silver<br/>dev_user_gold")]
    TEST --> TESTS[("test_user_bronze<br/>test_user_silver<br/>test_user_gold")]
    PRE -.-> PRES[("preprod_user_bronze<br/>preprod_user_silver<br/>preprod_user_gold")]

    DEVS --> SAFE["No shared tables.<br/>Nothing to break."]
    TESTS --> SAFE
    PRES -.-> SAFE

Loading

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:

presets:
  name_prefix: "[test ${workspace.current_user.short_name}] "

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 is worth doing

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.

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 preprod for full-load testing

Scaling this to preprod is a config block, not an architecture change:

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

Isolated targets are the foundation of Write-Audit-Publish (WAP) — the standard approach to not publishing bad data.

flowchart LR
    SRC["Incoming CSVs"] --> W["<b>WRITE</b><br/>pipeline runs into<br/>an isolated schema"]
    W --> A{"<b>AUDIT</b><br/>expectations +<br/>assertion job"}
    A -->|"passes"| P["<b>PUBLISH</b><br/>promote to<br/>production tables"]
    A -->|"fails"| STOP["Stop.<br/>Bad data never<br/>reaches consumers."]

Loading

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 test target runs the same pipeline against deliberately corrupted fixtures with 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_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 / test_gold, so it never touches dev.

See docs/testing.md for the defect table, expected counts, and known gaps.

Unit tests

The transformation logic lives in src/windmill/ as pure DataFrame transforms — no I/O, no dbutils, no session handling — so it runs against small local DataFrames without a cluster. 41 tests, ~11 seconds.

python3 -m venv .venv
.venv/bin/pip install -e ".[dev]"
.venv/bin/pytest

Requires a JDK (17 or later) for local Spark. The suite unsets SPARK_HOME before starting a session, so a Spark distribution already installed on the machine cannot cause a version clash.

The most important test is test_baseline_is_built_from_daily_means_not_raw_readings. It constructs data whose daily means are tightly clustered while the individual readings are spread wide, so the anomalous day is only detectable against the daily-mean baseline. It fails if the √24 defect is ever reintroduced — verified by injecting that defect and watching it fail, rather than assuming.

Cleanup

Remove everything the bundle created (schemas, volume, jobs, pipeline) for a target:

databricks bundle destroy --target dev
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.