microspace-planogram-platform

Batch + stream data pipelines (Airflow + Flink) that land item × club × fixture planogram data in a query-first Cassandra serving tier, exposed via Python FastAPI.

Part of the Enterprise Platform Reference Architecture.

Problem

Microspace / planogram data is a massive matrix: millions of item × club × fixture placements with intraday changes. Relational sharding breaks; you need pipeline orchestration, stream/batch processing, and Cassandra tables shaped around real queries.

Pipeline architecture

flowchart TB
  vendor[Vendor planogram CSV] --> airflow[Airflow DAGs]
  airflow --> minio[(MinIO bronze)]
  airflow --> pg[(Postgres staging + quarantine)]
  pg --> flink_batch[Flink batch job]
  changes[Space change CSV] --> flink_stream[Flink stream job]
  flink_batch --> cass[(Cassandra serving)]
  flink_stream --> cass
  cass --> api[FastAPI read API]

Loading

Tech stack (unique in portfolio)

Layer	Tech
Orchestration	Apache Airflow
Processing	Apache Flink (Java jobs)
Serving store	Apache Cassandra (TWCS history)
Bronze landing	MinIO (S3-compatible, no AWS)
Staging	PostgreSQL
Read API	Python FastAPI

No Kafka. No AWS services.

Cassandra tables

Table	Query
fixture_placement_by_club	All items on planogram X at club Y
item_placement_by_club_item	Where is item Z in club Y?
space_change_history	TWCS placement change timeline

Run

# Unit tests
mvn -f flink-jobs/pom.xml test
cd api && python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt && PYTHONPATH=. pytest -q tests

# Full stack
mvn -f flink-jobs/pom.xml package
docker compose up --build
./scripts/init-cassandra.sh
./scripts/demo.sh

Airflow UI: http://localhost:8080 (admin / admin)
Flink UI: http://localhost:8081
API: http://localhost:8089

Airflow DAGs

DAG	Purpose
ingest_vendor_planogram	Land CSV to MinIO bronze
validate_and_stage	Validate → Postgres staging / quarantine
promote_to_serving	Flink batch + stream → Cassandra

API

Method	Path
GET	/api/clubs/{clubNbr}/planograms/{planogramId}/fixtures
GET	/api/clubs/{clubNbr}/items/{itemNbr}/placement
GET	/api/clubs/{clubNbr}/items/{itemNbr}/history
POST	/api/admin/pipelines/trigger-ingest
POST	/api/admin/pipelines/trigger-promote

Documentation

• System design · erDiagram
• BRD · SOP · NFR · Cost savings
• ADRs: docs/adr/