Stage 2: package logic as a wheel with unit tests - #2
Open
p17b wants to merge 3 commits into
Open
Conversation
Extract the silver and gold logic from the notebooks into src/windmill/, built as a wheel by the bundle's artifacts block and attached to jobs. Notebooks reduce to orchestration: read, call a transform, write. Everything in the package is a pure DataFrame transform -- no I/O, no dbutils, no session handling -- which is what lets the layer be tested against small local DataFrames. 41 tests in ~11s, no cluster required. Serverless rejects a task-level `libraries` field, so dependencies are declared as a job `environments` block and referenced by environment_key from the task. deduplicate() now branches on df.isStreaming. dropDuplicatesWithinWatermark is rejected outright on batch DataFrames, and a watermark is meaningless there anyway since there is no unbounded state to bound; plain dropDuplicates on the same keys is equivalent. This means the unit tests exercise the production code path rather than a batch-only lookalike. The regression guard on the sqrt(24) defect was verified by reinjecting the defect rather than assumed. The first version of that test passed with the defect present -- it called turbine_baseline() directly, so it guarded the unit while the defect lived in how build_gold composed it. Correct components, wrong composition. The test now asserts through build_gold and fails as it should. Test fixtures pin an explicit schema instead of relying on inference: an all-null column in a small fixture has no inferrable type, and inference gave turbine_id a long where production has an int. conftest also drops SPARK_HOME before starting a session, so a Spark distribution already on the machine cannot cause a version clash -- a local 4.1.1 install against pip's 4.2.0 surfaced as an opaque "Method getConfs does not exist" py4j error. Verified behaviour-preserving against the test target: silver 11169, 54 invalid readings, 20 anomalies -- identical to the pre-refactor run. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Both branches deploy to the same targets, so a deployment carried no record of which code produced it. Tag every job with bundle.git.branch and bundle.git.commit so that is answerable from the Jobs UI. Tags rather than branch-derived names deliberately. name_prefix also applies to schema names, so deriving it from the branch would mean every branch switch renames the schemas -- a drop-and-recreate with data loss, for isolation that is not needed when branches are deployed one at a time. project tag moved to a top-level presets block shared by both targets; each target keeps its own environment tag and name_prefix. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Dedup keys on (timestamp, turbine_id) and gold takes an unweighted mean over readings. Both are correct only while the feed sends one reading per turbine per hour -- true of the provided data (all 11160 timestamps on :00:00, no duplicate keys, every one of the 465 turbine-days holding exactly 24 readings), but an observation rather than a contract. Sub-hourly timestamps would not break anything loudly. They are genuinely distinct readings, so dedup leaves them alone as it should, and the daily mean then quietly reweights toward whichever hours sent more samples. Gold now emits distinct_hours alongside measurement_count and sets has_sub_hourly_readings when they disagree, so the condition surfaces instead of silently distorting the average. No resampling rule is applied, because the correct one depends on what the sensor emits: instantaneous samples should be averaged, a cumulative counter needs the last value per hour, a pre-averaged period value needs weighting. The spec says megawatts -- a rate, not an energy total -- which points at the first case and makes averaging defensible, but that is inference rather than confirmation. Documented in the README with the options laid out, to be implemented once the sensor semantics are confirmed. Gold writes with overwriteSchema so adding a summary column no longer requires dropping the table by hand. Safe because gold is fully derived from silver and rebuilt from scratch every run. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Second of three stages, stacked on
medallion_workflow.Extracts the transformation logic from the notebooks into
src/windmill/, builtas a wheel by the bundle and attached to jobs. Notebooks reduce to orchestration:
read, call a transform, write.
Why
On stage 1 the logic lived inline in notebooks, which made it untestable without
a cluster. A full pipeline run costs minutes and needs a workspace — too slow a
loop for statistical logic, and exactly the kind of code where a subtle unit
mismatch hides (as the √24 defect on stage 1 demonstrated).
Everything in the package is a pure DataFrame transform — no I/O, no
dbutils,no session handling. That is what makes it testable: 41 tests in ~11 seconds,
no cluster.
Coverage
UNKNOWNfallback0/360boundaries that must stay validThe two test layers cover different things. Unit tests handle statistical
behaviour and edge cases awkward to express as fixtures; the stage 1 integration
harness covers what the package cannot see — Auto Loader, checkpoints, streaming
dedup across real re-runs, job wiring.
Mutation-checked
test_baseline_is_built_from_daily_means_not_raw_readingsguards the √24 defect.Rather than assume it worked, the defect was reinjected into
build_gold.It passed with the defect present. The test called
turbine_baseline()directly, so it guarded the unit while the defect lived in how
build_goldwiredit together — correct components, wrong composition, which is precisely the
failure mode unit tests are most prone to. The test now asserts through
build_goldand fails as it should.Notes
libraries; dependencies are declared as a jobenvironmentsblock referenced byenvironment_key.deduplicate()branches ondf.isStreaming.dropDuplicatesWithinWatermarkis rejected on batch DataFrames, and a watermark is meaningless there anyway —
so the unit tests exercise the production path rather than a lookalike.
conftestdropsSPARK_HOMEbefore starting a session. A local Spark 4.1.1install against pip's 4.2.0 surfaced as an opaque
Method getConfs does not existpy4j error; the suite is now hermetic.Verification
Behaviour-preserving against the test target: silver 11,169 / 54 invalid readings
/ 20 anomalies — identical to the pre-refactor run.