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

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

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.

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.

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.

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.

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

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

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.

Curate

Reconcile every store-day

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

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.

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

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

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.

Local — no cloud account

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

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.

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.

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:

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

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