Silent Data Loss Detector

A production-grade data engineering portfolio project that detects silent data loss between a source and destination database using Apache Spark, Apache Airflow, PostgreSQL, FastAPI, and Streamlit — all running locally via Docker Compose.

---

Why This Matters

Raw row counts alone cannot detect every form of data loss. If an ETL job silently duplicates 50,000 rows and drops a different 50,000, the row count looks identical. Only a checksum comparison reveals the corruption. This project demonstrates exactly that gap and gives you the full observability stack to catch it.

---

Architecture

┌────────────────────────────────────────────────────────────┐
│                      Docker Network                         │
│                                                             │
│  Parquet File                                               │
│  (mounted read-only)                                        │
│       │                                                     │
│       ▼                                                     │
│  ┌──────────┐    JDBC    ┌────────────────────────────┐    │
│  │  Spark   │ ─────────► │       PostgreSQL            │    │
│  │ (local   │ ◄───────── │  taxi_db                   │    │
│  │  mode)   │            │  ├── source.taxi_trips      │    │
│  └──────────┘            │  ├── destination.taxi_trips │    │
│       ▲                  │  └── reconciliation.results │    │
│       │ subprocess       │                            │    │
│  ┌──────────────┐        │  airflow (metadata DB)     │    │
│  │   Airflow    │        └────────────────────────────┘    │
│  │  Scheduler + │                  │                        │
│  │  Webserver   │                  │ psycopg2               │
│  └──────────────┘                  ▼                        │
│                           ┌─────────────┐                  │
│                           │   FastAPI   │                  │
│                           │  :8000      │                  │
│                           └──────┬──────┘                  │
│                                  │ HTTP                     │
│                           ┌──────▼──────┐                  │
│                           │  Streamlit  │                  │
│                           │  :8501      │                  │
│                           └─────────────┘                  │
└────────────────────────────────────────────────────────────┘

Data flow per DAG run:

1. Airflow triggers reconciliation_pipeline with a scenario parameter.
2. extract_metrics task calls ingest.py (Spark, local mode) to populate/refresh the source and destination tables, then calls compute_metrics.py to compute row counts, checksums, and null rates.
3. compute_checksums calculates deltas between source and destination.
4. compare evaluates whether row loss %, checksum mismatch, or null-rate delta exceeds configured thresholds.
5. write_results persists the full reconciliation record to PostgreSQL.
6. evaluate_threshold branches: FAIL → alert (Slack + email), PASS → log success.

---

Components

Service	Port	Description
PostgreSQL	5432	Three schemas: source, destination, reconciliation
Apache Spark	8080 (UI), 7077	Local-mode Spark for ingestion and metrics
Airflow Webserver	8090	DAG management UI
Airflow Scheduler	—	Runs the daily reconciliation pipeline
FastAPI	8000	REST API serving reconciliation results
Streamlit	8501	Two-page dashboard with auto-refresh

---

Running Locally

Prerequisites

• Docker Desktop (≥ 4.x) with at least 8 GB RAM allocated
• The Parquet file yellow_tripdata_2023-01.parquet must be in the project root

1. Configure environment

cp .env.example .env
# Edit .env — set SLACK_WEBHOOK_URL and SMTP_* values if you want real alerts
# The defaults work without alerts for local demo

2. Start all services

docker compose up --build -d

Wait approximately 2-3 minutes for Airflow to initialise (the airflow-init container runs airflow db migrate before the webserver starts).

3. Verify services are healthy

docker compose ps
# All services should show "running" or "healthy"

curl http://localhost:8000/health
# {"status":"ok","service":"silent-data-loss-detector-api"}

4. Trigger a DAG run

Open the Airflow UI at http://localhost:8090 and log in with admin / admin123.

In the DAG list, click reconciliation_pipeline, then click Trigger DAG w/ config and provide:

{ "scenario": "scenario_1" }

The first run triggers a full Spark ingestion of the 3 million row Parquet file (~5-10 min). Subsequent runs reuse the source table and only rebuild the destination.

5. View results

UI	URL
Streamlit Dashboard	http://localhost:8501
FastAPI Swagger	http://localhost:8000/docs
Airflow	http://localhost:8090
Spark Master UI	http://localhost:8080

---

Scenarios

Each scenario is triggered by setting the scenario param when manually triggering the DAG, or it defaults to scenario_1 on the daily schedule.

Scenario 1 — Row Drop (DOLocationID nulls)

{ "scenario": "scenario_1" }

Drops all destination rows where DOLocationID IS NULL.

Metric	Expected
Row loss	~2.8% (~85,000 rows)
Checksum	Mismatch
Result	FAIL — alert fired

Scenario 2 — Small Row Drop (fare_amount < $1)

{ "scenario": "scenario_2" }

Drops rows where fare_amount < 1.0.

Metric	Expected
Row loss	~0.4% (~12,000 rows)
Checksum	Mismatch
Result	PASS — below 5% threshold, no alert

Scenario 3 — Silent Duplicate Injection ⭐

{ "scenario": "scenario_3" }

This is the most important scenario. Rows 1–50,000 are replaced with exact duplicates of rows 50,001–100,000. Row count is identical to source. A naive row-count check says everything is fine. Only the checksum reveals that 50,000 records were silently corrupted.

Metric	Expected
Row loss	0.0000%
Row count diff	0
Checksum	MISMATCH
Result	FAIL — alert fired

Sample API response for a Scenario 3 FAIL run:

{
  "run_id": "a3f2c1d0-...",
  "scenario_name": "scenario_3",
  "overall_status": "FAIL",
  "checksum_mismatch": true,
  "source_row_count": 3066766,
  "destination_row_count": 3066766,
  "row_count_diff": 0,
  "row_loss_percentage": 0.0,
  "diverged_columns": []
}

The row count shows zero difference, yet the checksum catches that 50,000 rows were silently swapped for duplicates — exactly the data corruption that production pipelines miss.

---

API Reference

Endpoint	Description
GET /health	Service health
GET /runs	Last 30 reconciliation runs
GET /runs/{run_id}	Full detail for one run
GET /runs/{run_id}/diff	Column-level divergence for one run
GET /summary	Aggregated stats across all runs

---

Project Structure

.
├── docker-compose.yml
├── .env.example
├── yellow_tripdata_2023-01.parquet   ← mounted read-only into containers
├── airflow/
│   ├── Dockerfile                    ← extends apache/airflow:2.9.0 with Java + PySpark
│   └── dags/
│       └── reconciliation_pipeline.py
├── spark/
│   └── jobs/
│       ├── ingest.py                 ← Parquet → PostgreSQL + scenario transformation
│       └── compute_metrics.py        ← row counts, checksums, null rates
├── api/
│   ├── Dockerfile
│   ├── requirements.txt
│   ├── main.py
│   ├── models.py                     ← Pydantic schemas
│   ├── routes.py
│   └── db.py
├── dashboard/
│   ├── Dockerfile
│   ├── requirements.txt
│   └── app.py                        ← two-page Streamlit dashboard
└── postgres/
    └── init.sql                      ← schemas + reconciliation_results table

---

Stopping / Resetting

# Stop all services
docker compose down

# Stop and remove all volumes (full reset including DB data)
docker compose down -v

After down -v, the next docker compose up will re-ingest the source table from the Parquet file on the first DAG run.

---

Alert Configuration

Set the following in .env:

SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your@gmail.com
SMTP_PASSWORD=your-app-password
ALERT_EMAIL_TO=alerts@yourcompany.com

Alerts are idempotent — re-running a failed pipeline will not fire a duplicate alert for the same run_id.