TrainStream — Real-Time Railway Booking Analytics Pipeline

A real-time data pipeline simulating a railway booking platform (IRCTC-style) that generates two event streams — passenger searches and bookings — buffers them through Apache Kafka, processes them with PySpark Structured Streaming using windowed aggregations and watermarking for late-arriving data, and persists pre-aggregated metrics to PostgreSQL for downstream analytics and a live Streamlit dashboard.

---

Architecture

┌─────────────────────┐     ┌─────────────────────┐
│  search_producer.py │     │  booking_producer.py │
│  (Search Events)    │     │  (Booking Events)    │
└────────┬────────────┘     └──────────┬───────────┘
         │                             │
         ▼                             ▼
┌─────────────────────────────────────────────────┐
│               Apache Kafka (KRaft)              │
│         Topic: search_events                    │
│         Topic: booking_events                   │
└─────────────────────┬───────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────┐
│          PySpark Structured Streaming           │
│                                                 │
│  • JSON schema parsing                          │
│  • Watermarking (65 min late-data tolerance)    │
│  • 5-minute tumbling window aggregations        │
│  • foreachBatch → PostgreSQL upserts            │
└─────────────────────┬───────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────┐
│              PostgreSQL (Serving Layer)         │
│                                                 │
│  • bookings_per_route                           │
│  • status_breakdown                             │
│  • searches_per_route                           │
└─────────────────────┬───────────────────────────┘
                      │
                      ▼
┌─────────────────────────────────────────────────┐
│           Streamlit Dashboard (WIP)             │
└─────────────────────────────────────────────────┘

---

Tech Stack

Technology	Role	Why
Apache Kafka 4.1 (KRaft)	Message broker / event buffer	Decouples producers from consumers; persists events to disk so no data is lost if Spark goes down; multiple consumers can read the same topic independently at their own pace
PySpark 4.1 (Structured Streaming)	Stream processing engine	Parallelizes processing across cores; built-in DataFrame API with windowing, watermarking, and stateful aggregations out of the box — no need to implement these from scratch
PostgreSQL	Metrics serving layer	Solid JDBC support with Spark; native ON CONFLICT DO UPDATE upsert syntax for idempotent writes; good analytical query support for downstream BI
Docker	Container orchestration	Runs Kafka and PostgreSQL locally without native installs; volume mounts with SELinux :Z relabeling for Fedora compatibility
Python / confluent-kafka	Event generation	confluent-kafka (librdkafka wrapper) chosen over kafka-python for production-grade performance and reliability

---

Data Flow

1. search_producer.py continuously generates search events (user, route, travel class, quota) and writes them to the search_events Kafka topic.
2. booking_producer.py consumes search_events, and with 70% probability emits a corresponding booking event to booking_events. The booking event's timestamp is artificially offset by 5–60 minutes to simulate payment processing delay — creating realistic late-arriving data.
3. Spark reads both topics simultaneously via the Kafka source connector. Raw bytes are cast to strings and parsed into structured columns using from_json with explicit StructType schemas.
4. Watermarks of 65 minutes are applied to both streams — telling Spark to wait up to 65 minutes for late data before finalizing a window.
5. Three windowed aggregations are computed over 5-minute tumbling windows, triggered every 30 seconds.
6. Each micro-batch is written to PostgreSQL via foreachBatch using a temp table + INSERT ... ON CONFLICT DO UPDATE upsert pattern, ensuring each window has exactly one row that updates in place.
7. Spark checkpoints state to disk so the pipeline can resume from its last position after a restart without reprocessing.

---

Aggregations

bookings_per_route

Counts total bookings and seats requested per route per 5-minute window. Answers: Which routes are seeing the most demand right now?

status_breakdown

Counts bookings by status (CONFIRMED / WAITING / RAC) per 5-minute window. Answers: Are trains filling up? Is waitlist pressure increasing?

searches_per_route

Counts search events per route per 5-minute window. Combined with bookings_per_route in PostgreSQL, answers: What is the search-to-booking conversion rate per route?

SELECT
    b.window_start,
    b.route,
    b.total_bookings,
    s.total_searches,
    ROUND(b.total_bookings::numeric / NULLIF(s.total_searches, 0) * 100, 2) AS conversion_rate
FROM bookings_per_route b
JOIN searches_per_route s
    ON b.window_start = s.window_start
    AND b.route = s.route
ORDER BY b.window_start DESC;

---

Project Structure

TrainStreamData/
├── generators/
│   ├── search_events_producer.py    # Produces search events to Kafka
│   └── booking_events_producer.py   # Consumes searches, produces bookings
├── src/
│   ├── pipeline.py                  # PySpark Structured Streaming pipeline
│   ├── db.py                        # PostgreSQL connection pool
│   └── kafka_admin.py               # Topic auto-creation utility
├── jars/
│   └── postgresql-42.7.3.jar        # PostgreSQL JDBC driver (see setup)
├── checkpoints/                     # Spark checkpoint directories (git-ignored)
├── kafka-data/                      # Kafka persistence volume (git-ignored)
├── compose.yaml                     # Docker Compose (Kafka + PostgreSQL)
└── requirements.txt

---

Prerequisites

• Python 3.10+
• Java 17 (Spark 4.x does not support Java 25+)
• Docker + Docker Compose

---

Setup & Running

1. Clone and create virtual environment

git clone <repo-url>
cd TrainStreamData
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

2. Download PostgreSQL JDBC driver

mkdir -p jars
# Download postgresql-42.7.3.jar from https://jdbc.postgresql.org/download/
# and place it in the jars/ directory

3. Set environment variables

export PG_HOST=localhost
export PG_PORT=5433
export PG_DB=stream-db
export PG_USER=admin
export PG_PASSWORD=password
export JAVA_HOME=/usr/lib/jvm/java-17-temurin-jdk  # adjust for your system

4. Start infrastructure

docker compose up -d

5. Create PostgreSQL tables

docker exec -it train-db psql -U admin -d stream-db

CREATE TABLE bookings_per_route (
    window_start    TIMESTAMP,
    window_end      TIMESTAMP,
    route           TEXT,
    total_bookings  INTEGER,
    seats_requested INTEGER,
    CONSTRAINT uq_bookings UNIQUE (window_start, route)
);

CREATE TABLE status_breakdown (
    window_start    TIMESTAMP,
    window_end      TIMESTAMP,
    status          TEXT,
    booking_count   INTEGER,
    CONSTRAINT uq_status UNIQUE (window_start, status)
);

CREATE TABLE searches_per_route (
    window_start    TIMESTAMP,
    window_end      TIMESTAMP,
    route           TEXT,
    total_searches  INTEGER,
    CONSTRAINT uq_searches UNIQUE (window_start, route)
);

6. Run all three components (three separate terminals)

# Terminal 1
python generators/search_events_producer.py

# Terminal 2
python generators/booking_events_producer.py

# Terminal 3
python src/pipeline.py

---

Known Limitations

• Local only — runs on a single machine. A production deployment would use a distributed Kafka cluster and a Spark cluster (EMR, Databricks, or self-managed), which would surface additional distributed systems challenges not present here.
• Simplified event model — only two event types (search, booking). A real platform would include cancellations, seat assignments, berth preferences, PNR updates, and more — each potentially a separate Kafka topic.
• Unlimited seat assumption — the generator does not model seat inventory. Status (CONFIRMED/WAITING/RAC) is randomly assigned rather than derived from actual availability.
• At-least-once delivery — the pipeline does not implement exactly-once semantics. Duplicate events can occur if a producer restarts. The upsert logic in PostgreSQL mitigates the impact of duplicates on aggregated metrics.
• Event ID resets on producer restart — event_id counters are in-memory and reset to 1 on each producer restart. UUIDs would eliminate this limitation.

---

Future Improvements

• Distributed deployment — deploy Kafka and Spark on cloud infrastructure (EMR Serverless, Confluent Cloud) to handle real throughput and surface distributed systems challenges.
• Additional event streams — cancellations, seat inventory updates, PNR status changes as separate Kafka topics feeding into richer aggregations.
• ML layer — train a demand forecasting model on the aggregated metrics to predict which routes need additional services at what times.
• Schema Registry — use Confluent Schema Registry with Avro serialization instead of raw JSON, enforcing schema compatibility across producers and consumers.
• Exactly-once semantics — implement idempotent producers and transactional consumers for stronger delivery guarantees.
• Streamlit dashboard — live-updating visualization of the three aggregations reading directly from PostgreSQL.