# 🚀 Walmart Data Engineering Project | End-to-End AWS & Snowflake Pipeline

> An end-to-end retail pipeline built on synthetic data. Point-of-sale transactions in PostgreSQL and inventory in MongoDB are extracted by arrival time. Every store-day is reconciled against the store's close report, and daily sales and inventory KPIs are MERGEd into Snowflake. The local reference is verified by 12 passing tests; the AWS and Snowflake code is supplied but not yet run in the cloud.

- Source: https://akshaythakare.com/projects/walmart-retail-pipeline
- Stack: Python, PySpark, PostgreSQL, MongoDB, Docker, AWS Glue, Amazon S3, AWS Lambda, DynamoDB, Snowflake, Step Functions, Apache Airflow, Amazon SNS
- Tags: data-engineering, retail, aws, snowflake, pyspark, incremental-load, reconciliation, orchestration
- Updated: 2026-10-11

## Overview & the retail problem

## The problem

A large retailer's registers send transactions all day. Most reach the central database within minutes. Some don't: a register that lost its connection syncs the next morning, and the point-of-sale (POS) system sometimes sends the same line twice.

Store managers still need daily sales they can trust. Those sales must be:

- **complete**: every transaction rung up that day, including the late ones
- **not double-counted**: a resend must never add sales
- **reconciled**: equal to the store's own end-of-day close report

> This is an independent educational project with **synthetic data**. It is not Walmart's real infrastructure, data or code, and it is not sponsored by Walmart.

## What the pipeline does

1. Extract every row that **arrived** in a window, from PostgreSQL (POS) and MongoDB (inventory and promotions).
2. Validate the raw files against a manifest.
3. Merge them into curated history. A resend replaces the earlier copy.
4. Reconcile every store-day against the store close report.
5. Publish only the store-days that match. Facts are MERGEd into Snowflake at their grain.
6. Show sales, basket, department and stockout KPIs on a dashboard.

## The baseline you can check by hand

| Txn | Store | Lines | Net | Arrives |
| --- | --- | --- | --- | --- |
| T1 | 1001 East | 4 × $2.50 + 1 × $100.00 | $110.00 | same day |
| T2 | 1001 East | 2 × $4.00 (line resent once) | $8.00 | same day |
| T3 | 1002 West | $100.00 − $10.00 promotion | $90.00 | same day |
| T4 | 1002 West | return 1 × $2.50 | −$2.50 | same day |
| T5 | 1002 West | 3 × $4.00 | $12.00 | **next morning** |

**Window 1:** store 1001 publishes $118.00. Store 1002 has 2 of its 3 transactions, so it stays **INCOMPLETE**.

**Window 2:** T5 arrives and store 1002 publishes $99.50.

**Final result:** **$217.50 net sales · 5 transactions · 10 units · $43.50 average basket · 2.00 units per transaction**. Re-running either window changes nothing.

## Architecture & why each service

| Layer | Service | Why it is here |
| --- | --- | --- |
| Sources | PostgreSQL (POS), MongoDB (inventory) | Two real source shapes: relational transactions and document snapshots |
| Ingestion | AWS Glue | JDBC and MongoDB connectors; the window predicate is pushed down to PostgreSQL |
| Raw storage | Amazon S3 | Parquet partitioned by `business_date`, and a manifest per run |
| Validation | AWS Lambda | Fast checks of the manifest against S3 before Spark starts |
| Run ledger | DynamoDB | A conditional put stops the same window from being processed twice |
| Transformation | Glue PySpark | Merge into curated history, reconcile store-days, aggregate facts |
| Warehouse | Snowflake | COPY INTO staging, then MERGE in one transaction |
| Orchestration | Airflow → Step Functions | Airflow decides **when**; Step Functions decides **how** |
| Alerting | Amazon SNS | Every failed run sends its run id, window and error |

## Data plane and control plane

Files move through S3: `raw/` → `curated/` → `publish/`. Step Functions never carries data, only the `run_id` and the window.

The run id is a hash of the window, so a retried window maps to the same raw prefix and the same ledger item.

## What is verified

The local reference (`retail/`) implements every rule in Python and SQLite and is covered by 12 tests. The Glue, Lambda, Snowflake, Step Functions, Airflow and CloudFormation code mirrors it but has **not** been run in a cloud account yet.

## Source systems & synthetic data

## POS schema (PostgreSQL)

`stores`, `departments`, `products`, `transactions`, `transaction_lines` and `store_close`. Every fact table has an `ingested_at` column and an index on it, because extraction filters by arrival time.

`transaction_lines` deliberately has **no primary key**. Real POS feeds resend lines, so the pipeline must be able to deduplicate them.

## Inventory (MongoDB)

`inventory_snapshots` holds one document per store, SKU and day, with `on_hand` and `on_order`. `promotions` holds SKU lists with start and end dates and a discount.

## The generator

`retail/generate.py` is deterministic: the same seed always produces the same rows.

| Knob | Default | Example at scale |
| --- | --- | --- |
| `--stores` | 4 | 50 |
| `--days` | 3 | 30 |
| `--txns` (per store per day) | 40 | 400 |

It models a few retail patterns:
- Shoppers buy cheap staples far more often than electronics.
- Promotions lift basket quantity.
- About 3% of transactions are returns.
- About 2% of transactions sync the next morning.
- About 4% of shelves are empty at close.

Measured locally: 20 stores × 7 days × 300 transactions produces **42,000 transactions and 144,102 lines**, processed end to end in about **22 seconds**.

```bash
python3 -m retail.generate --stores 50 --days 30 --txns 400 --out data/generated
python scripts/load_sources.py --data data/generated     # into real PostgreSQL + MongoDB
```

## Incremental extract & raw validation

## Extract by arrival time

Each run covers a half-open window `[window_start, window_end)` on `ingested_at`. The dimension tables (stores, departments, products) are small, so they are snapshotted in full on every run.

Why not filter by business date? Suppose a sale rung up on Monday reaches the database on Tuesday. If Monday was already extracted, that sale would be missed forever. Arrival windows pick up every row exactly once, and the transform maps it back to its business date.

The run is refused when a window does not start at the stored **watermark**, so no gap can silently skip rows.

## The raw manifest

```text
raw/run_id=<sha256(window)[:16]>/
  table=transactions/part-*.parquet
  table=transaction_lines/…
  manifest.json      # rows, bytes and ETag (cloud) or SHA-256 (local) per table
```

## Validation (Lambda)

The validator checks:
- the manifest window matches the execution input
- all eight expected tables are present
- every file is inside the run prefix, and its size and ETag are unchanged
- required fields are present (checked locally)

It then claims the run in DynamoDB with `attribute_not_exists(run_id) OR status = FAILED`. A window that is in progress or has succeeded cannot be claimed again.

Two tests cover this: `test_corrupted_raw_file_fails_validation` and `test_window_must_start_at_watermark`.

## Reconciliation, late data & transformation

## Curate

| Rule | Implementation |
| --- | --- |
| Resends replace | Upsert on `(txn_id, line_no)`; the latest `ingested_at` wins |
| Bad lines are rejected | Unknown SKU, zero quantity or non-positive price go to `rejected_lines` |
| A correction clears the rejection | An accepted resend deletes the earlier rejection |
| Returns | Negative quantity, tracked separately as `return_cents` |
| Promotions | Line `discount_cents`, tracked separately |

## Reconcile every store-day

For every store-day this run touched, curated totals are compared with the store close report:

| Condition | Status | Effect |
| --- | --- | --- |
| No close report yet | `OPEN` | Nothing published |
| Count and net match, no rejected lines | `PUBLISHED` | Store-day MERGEd into the facts |
| Anything else | `INCOMPLETE` | Previously published facts stay; the day waits |

One store's late register never blocks the other stores.

## Facts

- `FACT_SALES_DAILY` has one row per date × store × department, with units, gross, discount, returns and net.
- `FACT_STORE_DAILY` has one row per date × store, with transactions, units and net.
- `FACT_INVENTORY_DAILY` has one row per date × store × SKU, with on hand, on order and a stockout flag.

In the cloud, the Glue job reads curated history only for the touched `business_date` partitions. It materialises the merge with `localCheckpoint()` before overwriting those partitions, because Spark cannot safely overwrite a path it is still lazily reading.

## Snowflake COPY INTO & MERGE

`02_load_merge.sql` runs per run id:

1. `COPY INTO` four temporary staging tables from `@PUBLISH_STAGE/run_id=…/`.
2. `BEGIN`.
3. Delete and re-insert `FACT_SALES_DAILY` rows for the republished store-days. A department that dropped to zero must disappear, which a plain MERGE would not do.
4. `MERGE` `FACT_STORE_DAILY`, `FACT_INVENTORY_DAILY`, `STORE_DAY_STATUS` and `PIPELINE_RUNS`.
5. `COMMIT`.

If anything fails before `COMMIT`, everything rolls back and the facts keep their previous values. This is single-database atomicity, not distributed exactly-once.

The publisher only accepts a run id of 16 hex characters before substituting it into the SQL.

## Dashboard views

| View | Shows |
| --- | --- |
| `KPI_DAILY` | Net sales, transactions, units, average basket, units per transaction |
| `SALES_BY_REGION` | Net sales by region and store format |
| `SALES_BY_DEPARTMENT` | Net, returns and discounts by department |
| `STOCKOUT_RATE` | Share of store-SKU snapshots at zero |
| `STORE_DAYS_PENDING` | Store-days waiting, with missing transactions and missing sales |

## Orchestration & alerting

## Airflow: when

`walmart_retail_daily` runs `@daily` with `catchup=True`, so a backfill replays the days one by one. `max_active_runs=1` keeps the windows in order. The DAG computes the window and the run id, starts Step Functions, and waits for it in reschedule mode.

## Step Functions: how

```text
ExtractRaw → ValidateBatch → TransformCurated → PublishSnowflake → MarkSucceeded
     ↘            ↘                 ↘                   ↘
                         MarkFailed → NotifyFailure (SNS) → Failed
```

- `ExtractRaw` retries Glue service errors twice.
- Every task's `Catch` routes to `MarkFailed`, which records the run in DynamoDB, then to `NotifyFailure`, which publishes run id, window and error.

## What is not a failure

A store-day waiting for late data is normal operation. It shows up in `STORE_DAYS_PENDING` and does not page anyone. A freshness alert for days that stay pending too long is the next step.

## Run it yourself

## Local — no cloud account

```bash
python3 -m unittest discover -s tests -t . -v      # 12 tests
python3 -m retail --root .runtime demo             # 4 daily windows, then KPIs
```

Expected output: the early windows report some `incomplete` store-days. After the last window, every store-day is `PUBLISHED`.

## The 12 tests

| Test | Proves |
| --- | --- |
| late transaction holds store-day | INCOMPLETE until T5 lands, then $217.50 |
| resent line counted once | Store 1001 stays at 7 units and $118.00 |
| returns and discounts | Gross, discount, returns and net per department |
| replaying a window | ALREADY_PROCESSED; facts identical |
| window must start at watermark | No gaps |
| failure before commit | Facts and watermark roll back; a retry succeeds |
| corrupted raw file | Checksum and missing-table detection |
| rejected line | Holds the day until a corrected resend |
| stockout rate | 1 of 4 snapshots at zero = 0.2500 |
| concurrent writer | A second writer is refused |
| generator deterministic | Same seed, same rows |
| generated data reconciles | Every store-day publishes and totals equal the close reports |

## Cloud

Follow `docs/STEPS.md` in the repository, steps 3–9: Docker sources, CloudFormation, Glue jobs, Snowflake setup, the state machine and the Airflow DAG. Each step ends with a check you can see.

## Verified vs planned & next steps

| Component | State |
| --- | --- |
| Generator, sources, extract, validation, reconciliation, MERGE, KPIs | **Tested locally** (12 tests) |
| PostgreSQL schema, Docker Compose, source loader | Written; schema mirrors the local tables |
| Glue extract and transform, validation Lambda, Snowflake publisher | Code supplied, **not run in AWS** |
| Snowflake setup, MERGE and views | Written, **not run against Snowflake** |
| Step Functions, Airflow, CloudFormation | Written, **not deployed** |
| Dashboard data export and preview | **Tested locally** (export: 9 tests; preview calculation rules: 22 tests) |
| Power BI report (`.pbix`), DAX measures | KPI views and `powerbi/measures.dax` written; **not built, DAX not run in Power BI Desktop** |
| Repository | Local only, **not public yet** |

## Next steps

1. Run the whole chain in a real account. Capture Glue run ids, Snowflake query ids and the reconciled KPIs as evidence.
2. Build the Power BI report from `powerbi/README.md` and pass its validation checklist.
3. Load test at 50 stores × 30 days. Record Glue timings and cost.
4. Add a freshness alert for store-days that stay INCOMPLETE past a deadline.
5. Publish the repository.

## Power BI dashboard & data usage

## Why a dashboard

The pipeline's job ends with published Snowflake tables. People only get value when they can read them. The report on top of those tables answers six kinds of questions, from six kinds of people:

| Who | Asks | Page |
| --- | --- | --- |
| Store manager | Did my store hit its numbers, and is yesterday final? | Stores |
| Merchandiser | Which departments carry sales, and what do discounts and returns do to net? | Executive overview |
| Regional manager | Which region is ahead, and which stores drag it down? | Executive overview, then Stores |
| Supply chain | Which stores are running out of stock? | Inventory & data quality |
| Finance analyst | Can I trust this number before I put it in a deck? | Inventory & data quality |
| Data engineer | Is the pipeline delivering complete data? | Inventory & data quality |

The interactive preview at the end of this page lets you try each of these questions with real pipeline output.

## What is behind each number

| Visual | Reads | Measure |
| --- | --- | --- |
| Net sales, sales by day, department and region | `FACT_SALES_DAILY` (date × store × department) | `Net Sales` |
| Transactions, average basket, units per transaction | `FACT_STORE_DAILY` (date × store) | `Transactions`, `Average Basket`, `Units per Transaction` |
| Stockout rate | `FACT_INVENTORY_DAILY` (date × store × SKU) | `Stockout Rate` |
| Data completeness, pending store-days | `STORE_DAY_STATUS` (date × store) | `Data Completeness`, `Pending Store-Days` |

Every measure is in `powerbi/measures.dax` and mirrors the Python and SQL logic that the 21 local tests cover, so the report reproduces the pipeline's numbers.

## The rule that makes the numbers trustworthy

A store-day reaches the sales facts only when its totals match the store's own close report. Until then it is **absent**, not zero. The dashboard says so in three places:

1. **Data completeness** is the share of store-days that are published.
2. The latest day is marked **provisional** while any of its store-days are pending.
3. A table lists each pending store-day with the transactions that have arrived against what the close report expects, how much of that day's sales is **not yet published**, and the **late shortfall** the pipeline is still waiting for.

In the demo data, after the last daily window 12 store-days are pending: for example one store has 148 of its 150 transactions in, and $11,420.60 of $11,600.10. After the catch-up window every store-day matches and net sales rise by exactly the close-report total of the store-days that were pending. A test asserts that equality.

## Two modelling rules worth knowing

- **A basket cannot be split by department.** One transaction contains items from many departments. When a department is selected, `Transactions`, `Average Basket` and `Units per Transaction` return blank, because showing a number would count the same basket once per department.
- **Net = gross − discounts − returns.** Returns are negative quantities; discounts are stored on the line. `Discount Rate` and `Return Rate` explain the gap between gross and net.

## Build it

1. Follow `powerbi/README.md`: connect to Snowflake, load the published tables (never the raw or curated layers), mark a date table, create the relationships.
2. Paste the measures from `powerbi/measures.dax`.
3. Build the three pages from the table in the guide.
4. Run the validation checklist before sharing the report. It compares the report with `RETAIL.MART.KPI_DAILY`, with the sum of the stores' close reports, and checks that a department selection blanks the basket measures.

To see the target before building, generate the sample data used by the preview: `python scripts/export_dashboard_data.py`.

## What is verified and what is not

| Part | State |
| --- | --- |
| Dashboard data export (`retail/dashboard.py`) | **Tested locally**: 9 tests, including reconciliation with the close reports |
| Preview on this page | Calculation rules tested against the exported data (22 tests); the DAX text it shows is checked against `measures.dax` |
| DAX measures | Written to match the tested logic; **not yet run in Power BI Desktop** |
| Power BI file (`.pbix`) and scheduled refresh | **Not built** |