Add a data pipeline: S3, Spark, PostgreSQL and Redshift

.dockerignore

@@ -11,6 +11,10 @@ data/raw/generated
 # The external dataset cache is 44 MB and downloads on demand.
 data/raw/external
 
+# The pipeline's lake: a 96 MB raw CSV and Spark's output. The pipeline
+# service mounts ./data instead, and baking it in copied it into every image.
+data/lake
+
 # Build outputs, not inputs.
 reports
 **/*.xlsx

.github/workflows/ci.yml

@@ -19,6 +19,26 @@ jobs:
         # the floor was the only version anyone had ever run.
         python-version: ["3.12", "3.13"]
 
+    # The warehouse tests compare every query on PostgreSQL with SQLite. Without
+    # a server they skip, so CI provides one rather than letting them pass by
+    # not running.
+    services:
+      postgres:
+        image: postgres:16
+        env:
+          POSTGRES_USER: rootsignal
+          POSTGRES_PASSWORD: rootsignal
+          POSTGRES_DB: rootsignal
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U rootsignal"
+          --health-interval 5s
+          --health-retries 10
+
+    env:
+      ROOTSIGNAL_WAREHOUSE_URL: postgresql://rootsignal:rootsignal@localhost:5432/rootsignal
+
     steps:
       - name: Checkout repository
         uses: actions/checkout@v5
@@ -29,14 +49,21 @@ jobs:
           python-version: ${{ matrix.python-version }}
           cache: pip
 
+      # Spark runs on the JVM; the conform job's tests need one.
+      - name: Set up Java
+        uses: actions/setup-java@v5
+        with:
+          distribution: temurin
+          java-version: "17"
+
       - name: Install project and development dependencies
-        run: python -m pip install -e ".[dev]"
+        run: python -m pip install -e ".[dev,pipeline]"
 
       - name: Generate controlled sample dataset
         run: python scripts/generate_sample_data.py --output-dir data/raw/generated
 
       - name: Run test suite
-        run: pytest
+        run: pytest -rs
 
       - name: Run Ruff checks
         run: ruff check .

.gitignore

@@ -40,3 +40,6 @@ mutants/
 .vscode/
 .idea/
 .DS_Store
+
+# The pipeline's local lake (rebuilt by scripts/run_pipeline.py)
+data/lake/

Dockerfile

@@ -22,13 +22,19 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
 
 WORKDIR /app
 
+# Spark runs on the JVM. The conform job is tested in here, on Linux, which is
+# also what Glue and EMR run it on.
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends openjdk-17-jre-headless \
+ && rm -rf /var/lib/apt/lists/*
+
 # Dependencies are installed from pyproject.toml alone, against a placeholder
 # package, so that editing source code does not invalidate the layer holding
 # pandas, numpy and streamlit.
 COPY pyproject.toml README.md ./
 RUN mkdir -p src/rootsignal \
  && touch src/rootsignal/__init__.py \
- && pip install --no-cache-dir ".[dev,dashboard,ai]" \
+ && pip install --no-cache-dir ".[dev,dashboard,ai,pipeline]" \
  && rm -rf src
 
 COPY . .

README.md

@@ -93,8 +93,10 @@ Reproduced by the committed test suite and evaluation scripts.
 | **Data quality** | **8 errors** detected across 10 raw tables; cleaning imputes 3 fields, drops 2 exact duplicates, quarantines 1 impossible line; **zero errors** after |
 | **Plan vs actual** | Attainment spans **87.4% to 125.6%** across 64 region/category/channel segments, evenly split 32 above plan and 32 below. KAM quota attainment runs **89.8% to 106.0%** across four managers, three ahead and one behind |
 | **Reproducibility** | Identical results on **numpy 1.26 under Windows and numpy 2.5 under Linux** — same signal, same impact to the cent, same forecast improvement. The container resolves the top of the declared dependency range rather than a lockfile, which is how the numpy 2.x break was found |
-| **Tests** | **316 automated tests**, passing on both dependency sets. One asserts that no output ever claims causation |
-| **Mutation testing** | `scripts/mutation_test.py` applies 37 operator and constant mutations to the KPI layer; the suite kills **34 (91.9%)**. The 3 survivors are equivalent mutants (`> 0` vs `>= 0` on a count that is never zero, or on a ratio where 0/0 is already NaN) |
+| **Data pipeline** | The same 1,067,371 lines landed in S3, conformed by a **PySpark** job, cleaned, and loaded into **PostgreSQL** (996,528 sales lines) and **Redshift Serverless** (996,531, loaded before stock codes were made case-insensitive). Spark's output is **identical to the pandas adapter's in all six tables**, compared column by column on the full dataset |
+| **Warehouse parity** | Every staging view and all seven analytical queries return the **same rows on PostgreSQL and Redshift as on SQLite**. The one tolerance is `aov`, which may differ by a cent on an exact half cent because the engines round ties differently |
+| **Tests** | **427 automated tests**, passing on both dependency sets. One asserts that no output ever claims causation |
+| **Mutation testing** | `scripts/mutation_test.py` applies 37 operator and constant mutations to the KPI layer; the suite kills **34 (91.9%)**. The 3 survivors are equivalent mutants (`> 0` vs `>= 0` on a count that is never zero, or on a ratio where 0/0 is already NaN). On the pipeline, `scripts/mutation_test_pipeline.py` applies **28 realistic bugs** (Spark's rounding, a lost CSV option, a dropped tie-breaker, a spending cap that only logs) and the suite catches **all 28**, up from 22 before the gaps it found were tested. Operator mutants on the pipeline code: **93 of 108** caught; the 15 survivors are wait times, output file counts and print formatting |
 
 **On that false-alarm figure.** The 12 counts every signal that matched nothing
 planted, wherever it was raised: 9 on the two control windows and 3 alongside
@@ -119,6 +121,13 @@ Or don't, and check the claims instead:
 docker build -t rootsignal . && docker run --rm rootsignal pytest
 ```
 
+Or run the whole pipeline on the real dataset, from download to warehouse:
+
+```bash
+docker compose up -d warehouse
+docker compose run --rm pipeline          # land, conform in Spark, load, publish
+```
+
 Then, in rough order of what is worth seeing first:
 
 ```bash
@@ -182,9 +191,35 @@ Evidence package  —  evidence, pattern, impact, confidence criteria, alternati
 Recommended investigation
 ```
 
-**Stack:** Python, pandas, NumPy, SQL (SQLite/PostgreSQL), Streamlit, Plotly,
-pytest, GitHub Actions, and an optional OpenAI-compatible LLM that the system
-works fully without. Power BI is deliberately out of scope.
+**Stack:** Python, pandas, NumPy, PySpark, SQL (SQLite, PostgreSQL, Amazon
+Redshift), AWS S3, Streamlit, Plotly, Power BI, pytest, Docker, GitHub Actions,
+and an optional OpenAI-compatible LLM that the system works fully without.
+
+### Data pipeline
+
+```text
+UCI workbook -> S3 raw zone -> Spark conform job -> S3 curated zone (Parquet)
+     -> cleaning + validation (nothing loads while an error remains)
+     -> PostgreSQL warehouse   + S3 clean zone -> Redshift Serverless (COPY)
+     -> report tables (rate/mix split, drivers, forecast accuracy) -> Power BI
+```
+
+`jobs/conform_online_retail.py` is a PySpark version of the Online Retail
+adapter with no imports from the package, so the same file runs locally, in the
+container and on a Spark cluster. `scripts/compare_spark_to_pandas.py` checks it
+against the pandas adapter on the full dataset. Cleaning stays in pandas: it
+owns the audit trail and the quarantine, and a second implementation would mean
+a second definition of every rule.
+
+Locally the lake is a folder and the warehouse a PostgreSQL container
+(`docker compose run --rm pipeline`). `scripts/run_pipeline_aws.py` runs the same
+stages with the lake in S3 and a Redshift Serverless warehouse, capped at 8 RPU
+with a usage limit that switches it off, and a teardown step that removes
+everything that bills. Glue and EMR Serverless subcommands exist for the Spark
+step but have not been run: the account used is not yet permitted Glue jobs,
+and its EMR Serverless quota had not taken effect. The Spark job itself has only
+run locally so far. [power_bi.md](docs/power_bi.md) describes the report on the
+warehouse.
 
 ## What's built
 
@@ -204,6 +239,8 @@ works fully without. Power BI is deliberately out of scope.
 | RootSignal engine | Evidence, pattern, impact, confidence and recommended investigation, ranked | [signals.md](docs/signals.md) |
 | Scenario evaluation | Measures whether the engine tells three planted situations apart and stays silent on two controls | [signals.md](docs/signals.md) |
 | SQL layer | Staging views, commercial mart and seven business queries, all executed by tests | [sql.md](docs/sql.md) |
+| Warehouse | The same SQL on PostgreSQL and Redshift, translated from one source and held row for row to SQLite | [power_bi.md](docs/power_bi.md) |
+| Pipeline | Raw, curated and clean zones on disk or in S3, a PySpark conform job, and report tables for BI | [power_bi.md](docs/power_bi.md) |
 | Excel reporting | Six operational workbooks, each opening with what its figures mean | [reporting.md](docs/reporting.md) |
 | Dashboard | Five Streamlit pages over a Streamlit-free, tested data layer | [dashboard.md](docs/dashboard.md) |
 | Presentation | One map from schema keys to language a business reader already has | [dashboard.md](docs/dashboard.md#the-words-on-the-screen) |
@@ -251,8 +288,10 @@ against six named criteria rather than a tuned score. It is validated against
 five planted scenarios — classifying three of three correctly at a medium
 confidence floor, with zero false alarms at a high one — and runs unchanged on
 a million real invoice lines from a public dataset, where it beats a naive
-forecast baseline by 38.8% across 49 backtest folds. Python, pandas, SQL,
-Streamlit; 308 tests, including one asserting that no output ever claims
-causation.
+forecast baseline by 38.8% across 49 backtest folds. That dataset moves through
+a pipeline from S3 through a PySpark job into PostgreSQL and Redshift, with
+Spark's output checked against the pandas implementation on every row. Python,
+pandas, PySpark, SQL, AWS, Streamlit; 427 tests, including one asserting that no
+output ever claims causation.
 
 </details>

docker-compose.yml

@@ -24,3 +24,38 @@ services:
     image: rootsignal
     profiles: ["tools"]
     command: pytest
+
+  # Source to warehouse: land the published data, conform it in Spark, load
+  # it into the warehouse below. The lake is ./data/lake on the host.
+  pipeline:
+    build: .
+    image: rootsignal
+    profiles: ["tools"]
+    depends_on:
+      warehouse:
+        condition: service_healthy
+    environment:
+      - ROOTSIGNAL_WAREHOUSE_URL=postgresql://rootsignal:rootsignal@warehouse:5432/rootsignal
+    volumes:
+      - ./data:/app/data
+    command: python scripts/run_pipeline.py
+
+  # The warehouse the pipeline loads into. Credentials are for a local
+  # container only; a deployed warehouse takes its URL from the environment.
+  warehouse:
+    image: postgres:16
+    environment:
+      - POSTGRES_USER=rootsignal
+      - POSTGRES_PASSWORD=rootsignal
+      - POSTGRES_DB=rootsignal
+    ports:
+      - "5433:5432"
+    volumes:
+      - warehouse-data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U rootsignal"]
+      interval: 5s
+      retries: 10
+
+volumes:
+  warehouse-data:

docs/docker.md

@@ -19,7 +19,7 @@ docker run --rm rootsignal pytest
 
 ```bash
 docker run --rm -p 8501:8501 rootsignal                 # the dashboard
-docker run --rm rootsignal pytest                       # all 308 tests
+docker run --rm rootsignal pytest                       # the full suite, Spark tests included
 docker run --rm rootsignal python scripts/detect_signals.py
 ```
 

docs/power_bi.md

@@ -0,0 +1,87 @@
+# Power BI on the RootSignal warehouse
+
+The pipeline loads the cleaned model and the analysis results into PostgreSQL,
+and this report reads them from there. It computes nothing the project has not
+already computed and tested: revenue comes from the fact table, and the country
+contributions, the rate and mix split, month-on-month change and forecast
+accuracy come from report tables the pipeline writes.
+
+| Overview | Drivers |
+| --- | --- |
+| ![Overview](screenshots/powerbi_overview.png) | ![Drivers](screenshots/powerbi_drivers.png) |
+| **Returns** | **Forecast** |
+| ![Returns](screenshots/powerbi_returns.png) | ![Forecast](screenshots/powerbi_forecast.png) |
+
+## Open it
+
+1. Fill the warehouse:
+   ```bash
+   docker compose up -d warehouse
+   docker compose run --rm pipeline
+   ```
+2. Open `docs/RootSignal.pbit` in Power BI Desktop. It is a template: the
+   report and model without data, so it asks for the connection and loads.
+3. Credentials: **Database**, user `rootsignal`, password `rootsignal`. These
+   belong to the local container in `docker-compose.yml`.
+4. The container has no TLS certificate, so Power BI asks to connect without
+   encryption: **OK**.
+
+Loading a million sales lines takes about a minute. On a machine with 8 GB of
+RAM, run the report and the Spark pipeline at different times; together they
+exhaust Docker's memory.
+
+## What is in it
+
+**Overview.** Net sales, invoices, average order value and units as cards, with
+a country slicer that filters them. Net sales by month, and a table of each
+month against the one before, taken from `rpt_monthly_trend`.
+
+**Drivers.** How much each country moved net sales between October and
+November 2011, from `rpt_country_contribution`.
+
+**Returns.** Why the return rate moved over the same two months, split into a
+rate effect (countries returning a different share) and a mix effect (demand
+moving toward countries that always return more), from
+`rpt_return_rate_components` and `rpt_return_rate_split`. The rate effect
+carries almost all of the movement.
+
+**Forecast.** Backtested error of each daily revenue model against the naive
+baseline, from `rpt_forecast_accuracy`. `seasonal_mean_7` is 38.8% better than
+naive over 49 folds, the figure the README reports.
+
+## Building it yourself
+
+The template is the reliable route. To rebuild the report from an empty file:
+
+- **Tables.** Get data → PostgreSQL database, server `localhost:5433`, database
+  `rootsignal`, Import. Tables appear with the schema as a prefix, as
+  `rootsignal fact_sales` and so on.
+- **Relationships.** `fact_sales` to `dim_date` on `date`, to `dim_customer` on
+  `customer_id` and to `dim_region` on `region_code`, each many to one, single
+  direction. Leave the `rpt_` tables unconnected; they are finished results.
+- **Products.** `fact_sales` to `dim_sku` on `sku_id` works too. Building the
+  report found 172 stock codes written in two cases or with a trailing space
+  (`15056BL` and `15056bl`); Power BI compares text without case and saw
+  duplicates. The pipeline now trims and upper-cases codes, so the relationship
+  is many to one like the others.
+- **Measures.** Power BI names ignore case, so a measure cannot be called
+  `Units` next to a column called `units`:
+  ```DAX
+  Net Sales = SUM ( 'rootsignal fact_sales'[net_sales] )
+  Total Units = SUM ( 'rootsignal fact_sales'[units] )
+  Invoices = DISTINCTCOUNT ( 'rootsignal fact_sales'[order_id] )
+  AOV = DIVIDE ( [Net Sales], [Invoices] )
+  ```
+  `Invoices` is a distinct count because one invoice lists several products.
+  Expected totals: £20,972,594 net sales, 40,077 invoices, 11,438,460 units,
+  £523.31 average order value.
+- **Month-on-month change** comes from `rpt_monthly_trend` rather than
+  `DATEADD`. `dim_date` holds only the days the retailer traded, and Power BI's
+  time-intelligence functions assume an unbroken calendar.
+- **Partial months** are already excluded from the monthly report tables; the
+  data stops on 9 December 2011 and the pipeline drops that month.
+- **Totals.** Turn off the total row on tables of rates and changes. Adding
+  percentages across rows produces a number that means nothing.
+
+A return rate is not a fill rate. Returns are customers sending goods back, not
+a warehouse failing to ship.