DEV Community: jayanti-neu

🚛 Freight Tracker: What I Did After Polishing the UI

jayanti-neu — Mon, 22 Sep 2025 18:51:59 +0000

After finally making all the pages in the Freight Tracker frontend look more polished — with a cleaner layout, better navigation, nicer styling, and a responsive design — I thought I was done. But I wasn't. There was still a lot to improve under the hood.

Here's a breakdown of everything I worked on after the UI was "done," and what I learned along the way:

🔄 Making the App Feel More Dynamic with Zustand and WebSockets

I had already integrated Zustand in Phase 2, but now I made it more useful. Instead of hitting the API every time, I could cache and reuse shipments in the Zustand store. That made the app snappier.

Then I added WebSocket support to make live updates possible — so if a shipment gets updated in real-time (e.g., status changes to "DELIVERED"), it reflects instantly in the UI without the user refreshing the page. This gave the app a much more modern feel.

What I learned:

Zustand is great for global state without the boilerplate of Redux
useShipmentStore.getState() gives access outside components
WebSockets with STOMP and SockJS were already set up in the backend, so I just needed to connect and subscribe to the right topic
Having WebSockets + Zustand together means my data stays fresh and fast

🗺️ Making the Map More Realistic

The Map Dashboard looked okay, but loading was slow. Why? Because I was calling the Google Geocode API for every city — every time.

So I added a geocode cache using localStorage. Now if a city was already looked up before, we reuse the coordinates. This made route rendering way faster.

I also:

Gave each route a different color based on shipment status
Added loading indicators while geocoding
Made the map update live when new shipments are added via WebSocket

⚠️ Fixing the Refresh Issue on Netlify

One of the most annoying bugs I ran into: when I refreshed a route like /shipments/2, I got a "page not found" error. Turns out this is a common issue with React apps deployed on Netlify.

The fix? A simple _redirects file in the public/ folder:

/* /index.html 200
That tells Netlify to send all routes to index.html so React Router can do its thing.

🕒 Preventing Backend Sleep with GitHub Actions

After deploying the backend to Render, I noticed that the free tier web service would go to sleep after a few minutes of inactivity. This meant that every time I revisited the app after a while, the backend had to "wake up," causing slow response times or even temporary unavailability.

To fix this, I added a lightweight GitHub Actions workflow in my backend repository that pings the backend url every 10 minutes — just enough to keep it alive without overwhelming it.

Here's what the workflow looks like:


yaml
# .github/workflows/keep-alive.yml
name: Keep Render Backend Alive

on:
  schedule:
    - cron: '*/10 * * * *'  # runs every 10 minutes
  workflow_dispatch:        # allows manual triggering

jobs:
  ping:
    runs-on: ubuntu-latest
    steps:
      - name: Curl the backend
        run: curl --s https://freight-tracker.onrender.com/api/shipments

Once pushed to the main branch, it showed up in the Actions tab on GitHub. Now I never have to worry about the backend sleeping on me. 🚀
⚠️ Note: GitHub Actions schedules are “best-effort” and may not always run exactly on time. If you want a more consistent keep-alive mechanism, consider using a free external ping service like UptimeRobot
 or cron-job.org


## 🌍 Frontend + Backend Deployment

The frontend was deployed on **Netlify** and the backend (Spring Boot) was containerized and deployed on **Render** using Docker.



Key steps I learned:
* Backend needs CORS configured correctly (e.g. allow frontend URL)
* Set up a PostgreSQL DB on Render and connect via environment variables
* Render's default `.mvnw` build failed, so I switched to Docker and wrote a `Dockerfile` instead
* Frontend needed environment variable for backend API (`VITE_API_URL`)
* I had to commit my real `application.properties`, not `application.properties.example`, with placeholders like `${DB_URL}` — and set those secrets in Render





Here's the live app: https://freighttracker.netlify.app/

🚛 Freight Tracker Frontend — Phase 2 & Phase 3 Recap

jayanti-neu — Mon, 25 Aug 2025 21:48:24 +0000

React + TypeScript + TailwindCSS + WebSockets + Zustand + Leaflet.js

🧭 Overview

The frontend of Freight Tracker has come a long way! In Phase 2 and Phase 3, we shifted from a static layout to a dynamic, interactive logistics dashboard powered by live updates and rich UI components. In this post, I'll walk through the main features implemented, how we handled state management and live updates, and some challenges I encountered (and solved).

📦 Phase 2: CRUD Pages, State Management & Styling

✅ 1. Shipment Type Definitions

We created proper TypeScript types (Shipment, ShipmentStatus, ShipmentStatsDTO, etc.) to mirror the backend schema and ensure type-safe code across components.

export type ShipmentStatus = "PENDING" | "IN_TRANSIT" | "DELIVERED" | "CANCELLED";

export interface Shipment {
  id: number;
  origin: string;
  destination: string;
  status: ShipmentStatus;
  trackingNumber: string;
  carrier?: string;
  priority?: "LOW" | "MEDIUM" | "HIGH";
  lastUpdatedTime: string;
}

🧠 2. Global State with Zustand

Zustand was introduced to maintain and manipulate shipment data globally. This allowed pages like the list, details, and map view to share consistent data without prop drilling.

const useShipmentStore = create((set) => ({
  byId: {},
  upsert: (shipment) => set((state) => ({
    byId: { ...state.byId, [shipment.id]: shipment }
  })),
  bulkLoad: (shipments) => set({
    byId: Object.fromEntries(shipments.map(s => [s.id, s]))
  }),
}));

🖼️ 3. Basic Pages with Routing

We added multiple pages using react-router-dom:

/ — HomePage
/shipments — ShipmentsPage
/shipments/:id — ShipmentDetailsPage
/dashboard — MapDashboardPage (added in Phase 3)

Routing was kept simple, with a layout container and top navigation.

🖍️ 4. Styling with TailwindCSS

All pages were styled with Tailwind for quick, consistent design. We used colors, spacing, typography, and component states (like animate-pulse on loading) to make the UI pleasant and professional.

⚡ Phase 3: WebSocket Updates, Maps, and Visualization

🌐 1. Real-Time WebSocket Integration

Using @stomp/stompjs and a WebSocket backend endpoint (/ws), we added live updates to shipments. Any change to a shipment in the backend triggers a broadcast message, which is parsed and upserts the new shipment into Zustand.

stompClient?.subscribe("/topic/shipments", (message) => {
  const shipment: Shipment = JSON.parse(message.body);
  useShipmentStore.getState().upsert(shipment);
});

This ensures all views (including the map) reflect changes instantly — no need to refresh!

🗺️ 2. Interactive Map with Leaflet

We created a rich dashboard map to visualize routes between origin and destination cities. Highlights:

Leaflet.js with react-leaflet for the map
Geocoding city names using OpenCage API
Route lines colored based on shipment status
Markers with popups for extra detail

<Polyline positions={[originCoords, destinationCoords]} color={getColorForStatus(shipment.status)}>
  <Popup>
    <strong>Tracking:</strong> {shipment.trackingNumber}
    <br />
    <strong>Status:</strong> {shipment.status}
  </Popup>
</Polyline>

We also cached geocode results in localStorage to avoid repeated lookups.

💡 3. State-Driven Map Updates

Instead of fetching shipment data directly in the map page, we relied on Zustand. However, we included a fallback to getShipments() on first load in case the store was still empty. We watched Object.values(byId).length to trigger re-renders when shipments were updated.

useEffect(() => {
  const shipments = Object.values(byId);
  if (shipments.length === 0) {
    const fresh = await getShipments();
    bulkLoad(fresh);
  }
  ...
}, [byId]);

This worked hand-in-hand with WebSocket updates — ensuring the map reflects changes as soon as they happen.

🚦 4. Small Optimizations

Loading indicator while routes are geocoded
Conditional polyline coloring
Map centering and zoom configuration
Clean Leaflet icon setup
Using React.Fragment with stable key props for rendering

🧠 Lessons Learned

Zustand is great for lightweight state management with zero boilerplate
WebSocket integration with React is smooth with STOMP + Zustand
It's important to debounce or cache expensive calls like geocoding
Use localStorage smartly to persist data across sessions
Routing doesn't cause component remounts by default — understand how effects work in SPA frameworks like React!

📌 Next Steps

Add authentication and role-based views (client/admin separation)
Real-time truck location tracking
Better filtering and sorting options
UI polish and animations
Backend caching for geocodes (to avoid using frontend localStorage in production)

🚀 Building the Freight Tracker Frontend (Phase 1)

jayanti-neu — Thu, 31 Jul 2025 15:13:29 +0000

In the previous phases, we built the backend API and added real-time updates with WebSockets. Now it's time to bring the frontend to life — building a React + TypeScript + Tailwind Single Page Application (SPA) that talks to our backend.

This post covers Phase 1 of the frontend: setting up the project, defining its folder structure, and configuring routing for multiple pages.

🎯 Goals of Phase 1

Set up React project with Vite + TypeScript + Tailwind CSS
Create a scalable folder structure used in professional projects
Implement basic routing (Home + Shipments pages)
Understand SPA architecture and why React Router is needed

🛠 Tech Stack

React (UI library)
TypeScript (type safety)
Vite (fast dev bundler)
Tailwind CSS (utility-first styling)
React Router DOM (SPA routing)

1. Project Setup

Create Vite project

npm create vite@latest client-freight
# Choose React + TypeScript
cd client-freight

Install Tailwind (latest)

npm install -D tailwindcss @tailwindcss/vite

Edit vite.config.ts:

import { defineConfig } from "vite";
import react from "@vitejs/plugin-react";
import tailwindcss from "@tailwindcss/vite";

export default defineConfig({
  plugins: [react(), tailwindcss()],
});

Add Tailwind directives in index.css:

@tailwind base;
@tailwind components;
@tailwind utilities;

Run:

npm run dev

You should see the default Vite React page at http://localhost:5173.

2. Professional Folder Structure

Here's the structure we created under src/:

src/
│
├── components/        # Reusable UI components (buttons, cards, etc.)
├── pages/             # Full-page components (HomePage, ShipmentsPage)
├── hooks/             # Custom React hooks (e.g., useShipments)
├── services/          # API calls / WebSocket helpers
├── types/             # TypeScript types/interfaces
├── context/           # React context (e.g., AuthContext, ThemeContext)
├── App.tsx            # Main app component (routing setup)
├── main.tsx           # Vite entry point
└── index.css          # Tailwind global styles

Why this structure?

components/: Small reusable building blocks (e.g., <Button />, <Card />)
pages/: Represents real screens/routes (/ for home, /shipments for shipments list)
hooks/: Encapsulate logic like data fetching or WebSocket subscriptions
services/: All HTTP/WebSocket calls centralized here (e.g., getShipments())
types/: Centralized TypeScript interfaces to keep backend and frontend consistent
context/: Global state management (e.g., authentication, theme)

This separation scales well as the project grows and keeps responsibilities clear.

3. What is an SPA and Why React Router?

How SPAs Work

A Single Page Application (SPA) loads one HTML page and dynamically updates the content with JavaScript (React) instead of requesting new HTML pages from the server.

Faster navigation after first load
No full-page refresh — feels more like a native app
Better user experience for dashboards, real-time apps, etc.

Why React Router DOM?

You could theoretically build a single-view SPA with just React and no router at all. React Router DOM becomes necessary when you want that SPA to feel like it has multiple "pages" that users can navigate between.

Without React Router, you'd have to manually manage window.location and conditional rendering. React Router DOM handles:

URL-based navigation (/, /shipments)
Back/forward browser buttons
Page transitions without reloads
Preserving state between page views

But React itself is what makes SPAs possible by:

Managing the virtual DOM and efficiently updating only changed parts
Creating reusable components that can be dynamically rendered
Handling state changes without page reloads
Providing the foundation for building interactive UIs

4. Implementing Routing

Install React Router DOM:

npm install react-router-dom

App.tsx

import { BrowserRouter as Router, Routes, Route, Link } from "react-router-dom";
import HomePage from "./pages/HomePage";
import ShipmentsPage from "./pages/ShipmentsPage";

export default function App() {
  return (
    <Router>
      <nav className="p-4 bg-gray-800 text-white flex gap-4">
        <Link to="/">Home</Link>
        <Link to="/shipments">Shipments</Link>
      </nav>

      <Routes>
        <Route path="/" element={<HomePage />} />
        <Route path="/shipments" element={<ShipmentsPage />} />
      </Routes>
    </Router>
  );
}

pages/HomePage.tsx

export default function HomePage() {
  return <h1 className="text-2xl font-bold p-4">Welcome to Freight Tracker</h1>;
}

pages/ShipmentsPage.tsx

export default function ShipmentsPage() {
  return <h1 className="text-2xl font-bold p-4">Shipments</h1>;
}

How Navigation Works

User visits / → Sees nav bar + HomePage
Clicks "Shipments" link → URL changes to /shipments, React swaps HomePage with ShipmentsPage (no reload)
Nav bar stays persistent across all routes

This is what makes SPAs fast and seamless.

5. Key Learnings from Phase 1

Tailwind + Vite gives rapid dev speed + modern CSS utilities
SPA model: one-page load, client-side routing, smooth navigation
React Router DOM: essential for multiple "pages" in SPAs
Folder structure: separation of pages, components, services, types keeps things clean

6. What's Next?

Phase 2 Frontend: Fetch shipments from backend API
Add real-time WebSocket updates
Create responsive UI components (cards, tables)
Later: auth, filtering, deployment

GitHub Repo

Frontend code is available here: GitHub - freight-tracker-frontend

🚚 Building a Real-Time Freight Tracker in Java (Phase 3)

jayanti-neu — Tue, 29 Jul 2025 22:29:33 +0000

In Phase 1, we built core CRUD APIs and statistics endpoints.
In Phase 2, we introduced real-time WebSocket updates so clients see shipment changes instantly.

Phase 3 is about refactoring and hardening the backend:

Adding a Service Layer to cleanly separate logic
Centralizing error handling with @ControllerAdvice
Using DTOs + @Valid for safe and validated input
Writing unit and integration tests
General cleanup to prepare for production

🧠 Why Refactor?

As features grow, putting everything in controllers (like we did earlier) becomes messy. Controllers shouldn't know about database details or WebSocket internals. Instead, they should delegate to a service layer. This also makes testing easier and avoids duplication.

1️⃣ Service Layer

Why create a service interface?

Defines a contract — what operations are available
Easier to test — we can mock the service in controller tests
Future flexibility — switch implementations without breaking controllers

ShipmentService.java

public interface ShipmentService {
    Shipment createShipment(CreateShipmentRequest request);
    Shipment getShipmentByIdOrThrow(Long id);
    List<Shipment> getAllShipments();
    Shipment updateShipment(Long id, UpdateShipmentRequest request);
    void deleteShipment(Long id);
    Page<Shipment> searchShipments(String origin, ShipmentStatus status, Pageable pageable);
    ShipmentStatsDTO getShipmentStats();
}

ShipmentServiceImpl.java

@Service
public class ShipmentServiceImpl implements ShipmentService {

    @Autowired
    private ShipmentRepository shipmentRepository;

    @Autowired
    private ShipmentStatusBroadcaster broadcaster;

    @Override
    public Shipment createShipment(CreateShipmentRequest request) {
        Shipment shipment = new Shipment();
        shipment.setOrigin(request.getOrigin());
        shipment.setDestination(request.getDestination());
        shipment.setStatus(request.getStatus());
        shipment.setTrackingNumber(request.getTrackingNumber());
        shipment.setCarrier(request.getCarrier());
        shipment.setPriority(request.getPriority());
        shipment.setLastUpdatedTime(LocalDateTime.now());

        Shipment saved = shipmentRepository.save(shipment);

        broadcaster.broadcastUpdate(
            ShipmentUpdateMessage.builder()
                .shipmentId(saved.getId())
                .trackingNumber(saved.getTrackingNumber())
                .status(saved.getStatus())
                .lastUpdatedTime(format(saved.getLastUpdatedTime()))
                .build()
        );

        return saved;
    }

    // Other methods like updateShipment, getShipmentByIdOrThrow, etc.
}

Key takeaway:

Controller only handles routing and leaves all logic to service
Service calls repository and broadcaster, making code modular and testable

2️⃣ Centralized Error Handling

Before: Controllers used try/catch blocks or returned ResponseEntity.notFound() directly. This caused repetition and inconsistent error messages.

After: We created a global handler using @ControllerAdvice:

@ControllerAdvice
public class GlobalExceptionHandler {

    @ExceptionHandler(ShipmentNotFoundException.class)
    public ResponseEntity<Map<String, Object>> handleNotFound(ShipmentNotFoundException ex) {
        Map<String, Object> errorBody = new HashMap<>();
        errorBody.put("status", "error");
        errorBody.put("message", ex.getMessage());
        errorBody.put("timestamp", LocalDateTime.now());
        return new ResponseEntity<>(errorBody, HttpStatus.NOT_FOUND);
    }

    @ExceptionHandler(MethodArgumentNotValidException.class)
    public ResponseEntity<Map<String, Object>> handleValidationErrors(MethodArgumentNotValidException ex) {
        Map<String, String> fieldErrors = ex.getBindingResult().getFieldErrors().stream()
                .collect(Collectors.toMap(FieldError::getField, FieldError::getDefaultMessage));

        Map<String, Object> response = new HashMap<>();
        response.put("status", "error");
        response.put("errors", fieldErrors);
        response.put("timestamp", LocalDateTime.now());

        return new ResponseEntity<>(response, HttpStatus.BAD_REQUEST);
    }
}

Why this is better:

One place handles all exceptions
Clean controllers (no try/catch spam)
Consistent JSON error format for frontend

3️⃣ DTOs + @valid

DTO (Data Transfer Object)

Controls what fields the client can send/receive
Prevents exposing internal database fields (id, lastUpdatedTime)
Allows different validation rules for create vs update requests

Example: CreateShipmentRequest

@Data
public class CreateShipmentRequest {
    @NotBlank(message = "Origin is required")
    private String origin;

    @NotBlank(message = "Destination is required")
    private String destination;

    @NotNull(message = "Status is required")
    private ShipmentStatus status;

    @NotBlank(message = "Tracking number is required")
    @Size(min = 5, message = "Tracking number must be at least 5 characters long")
    private String trackingNumber;

    private String carrier;
    private Priority priority;
}

@valid

Automatically validates DTO fields before calling the service
If invalid, throws MethodArgumentNotValidException (caught by our global handler)
Removes the need for manual if checks in controllers

Difference Between DTO and @valid

DTO: Shapes the data (what fields are allowed)
@valid: Enforces rules on that data (what values are allowed)

Together:

DTO ensures we only accept specific fields
@valid ensures those fields have valid values

4️⃣ Testing

We wrote unit tests for the service layer and integration tests for the controller.

Unit Test Example (Service Layer)

@Test
void testCreateShipment() {
    // Simulate DB behavior
    when(shipmentRepository.save(any(Shipment.class))).thenAnswer(invocation -> {
        Shipment s = invocation.getArgument(0);
        s.setId(1L); // Simulate DB assigning ID
        return s;
    });

    CreateShipmentRequest req = new CreateShipmentRequest();
    req.setOrigin("NY");
    req.setDestination("Chicago");
    req.setStatus(ShipmentStatus.IN_TRANSIT);
    req.setTrackingNumber("TRK12345");

    Shipment result = shipmentService.createShipment(req);

    assertEquals("NY", result.getOrigin());               // Verify service mapped correctly
    verify(broadcaster, times(1)).broadcastUpdate(any()); // WebSocket should be triggered
}

Integration Test Example (Controller + H2 DB)

@Test
void createShipment_andGetAllShipments() throws Exception {
    CreateShipmentRequest request = new CreateShipmentRequest();
    request.setOrigin("New York");
    request.setDestination("Chicago");
    request.setStatus(ShipmentStatus.IN_TRANSIT);
    request.setTrackingNumber("TRK12345");

    // POST request
    mockMvc.perform(post("/api/shipments")
            .contentType(MediaType.APPLICATION_JSON)
            .content(objectMapper.writeValueAsString(request)))
        .andExpected(status().isOk())
        .andExpected(jsonPath("$.origin").value("New York"));

    // GET request
    mockMvc.perform(get("/api/shipments"))
        .andExpected(status().isOk())
        .andExpected(jsonPath("$[0].origin").value("New York"));
}

Why both Unit and Integration?

Unit tests check business logic in isolation (fast, targeted)
Integration tests verify the full request → controller → service → DB flow
Both together give confidence: logic works and wiring works

🗓️ What's Next?

Frontend (React): Build a dashboard for clients and admins
Authentication: Basic login for admins to manage shipments
Cloud Deployment: Docker + AWS (EC2 or RDS)
Advanced Features: More analytics, filters, and push notifications

👩‍💻 Final Thoughts

Phase 3 transformed our backend from a basic prototype into a clean, testable, production-ready API. By introducing service layers, DTOs, validations, and tests, we've laid the foundation for scaling features without chaos.

You can check the repo here: https://github.com/jayanti-neu/freight-tracker

Building a Real-Time Healthcare Data Pipeline with Apache Spark: From SQS to Parquet (Part 2)

jayanti-neu — Sun, 20 Jul 2025 15:41:42 +0000

Building a Real-Time Healthcare Data Pipeline - Phase 2: Stream Processing with Apache Spark

This is Part 2 of my VitalWatch series. In Phase 1, I built a producer that streams ICU patient data from MIMIC-IV to AWS SQS. Now I'm building the processing engine that transforms this raw stream into analytics-ready data.

The Evolution: From Message Queue to Data Lake

Where We Left Off

Phase 1 created a data faucet - ICU patient data flowing continuously into an SQS queue. But raw messages sitting in a queue aren't useful for healthcare analytics. We need to process, structure, and store this data efficiently.

The Big Picture: What Phase 2 Accomplishes

Phase 1: ICU Data → SQS Queue
Phase 2: SQS Queue → Spark → Analytics → Parquet Files → S3
Phase 3: S3 → Dashboard → Alerts [Coming Next]

Phase 2 is like adding a smart water treatment plant that:

Continuously reads from the SQS queue as messages arrive
Processes the raw JSON data into structured formats
Transforms data (timestamps, data validation, schema enforcement)
Stores results in an organized, queryable format

Understanding the Technology Stack - Key Technologies Explained

Why Apache Spark for Stream Processing?

Apache Spark is like a super-powered Excel that can process millions of rows in real-time. Unlike traditional databases that struggle with large-scale analytics, Spark excels at:

Distributed processing: Spreads work across multiple CPU cores (or machines)
In-memory computing: Keeps data in RAM for faster processing
Fault tolerance: Automatically recovers from failures using lineage information
Schema-on-read: Applies structure to data at query time, not storage time

Why Spark works better for analytics than traditional databases:
Traditional databases maintain ACID properties (Atomicity, Consistency, Isolation, Durability), which requires complex coordination between nodes in distributed systems. Spark trades off some consistency guarantees for massive performance gains in analytics workloads where eventual consistency is acceptable.

The Storage Revolution: Parquet Format

Moving from CSV to Parquet is like upgrading from a filing cabinet to a modern database:

CSV (Row-oriented):

patient_id,age,diagnosis,length_of_stay
1,25,Heart Attack,5.2
2,67,Pneumonia,8.7

Parquet (Column-oriented):

patient_id: [1, 2, 3, ...]
age:        [25, 67, 45, ...]
diagnosis:  [Heart Attack, Pneumonia, ...]
los_hours:  [5.2, 8.7, 3.1, ...]

The analytics advantage: When you query "What's the average length of stay?", Parquet only reads the los_hours column, making queries 10x faster and using much less storage due to column-specific compression algorithms.

Architecture Deep Dive

Data Schema Definition

schema = StructType([
    StructField("subject_id", IntegerType()),    # Patient identifier
    StructField("stay_id",    IntegerType()),    # ICU stay identifier
    StructField("intime",     StringType()),     # Admission time (converted later)
    StructField("outtime",    StringType()),     # Discharge time (converted later)
    StructField("los_hrs",    DoubleType()),     # Length of stay in hours
])

Why explicit schemas matter:

Performance: Spark doesn't need to infer data types by scanning data
Data quality: Type validation catches malformed data at ingestion time
Consistency: Ensures all batches have the same structure across time
Memory efficiency: Proper types use optimal memory allocation

The Stream Processing Pipeline

def next_batch():
    messages = []
    while not messages:
        resp = sqs.receive_message(
            QueueUrl=QUEUE_URL,
            MaxNumberOfMessages=min(BATCH_SIZE, 10),  # SQS limit
            WaitTimeSeconds=10,
        )
        messages = resp.get("Messages", [])

    # Clean up processed messages to prevent reprocessing
    entries = [{"Id": m["MessageId"], "ReceiptHandle": m["ReceiptHandle"]} for m in messages]
    if entries:
        sqs.delete_message_batch(QueueUrl=QUEUE_URL, Entries=entries)

    return [m["Body"] for m in messages]

Key Design Decisions Explained

1. Batch Processing with SQS Limits

MaxNumberOfMessages=min(BATCH_SIZE, 10)

The constraint: AWS SQS has a hard limit of 10 messages per request. Even if I set BATCH_SIZE=50, each batch actually processes a maximum of 10 messages per next_batch() call.

Current limitation: To truly process 50 messages per batch, the code would need to accumulate multiple SQS calls within the next_batch() function.

Why this matters:

Keeps processing efficient without overwhelming the system
Respects AWS service limits
Spark performs better with larger batches, but we're currently limited by this design

2. Long Polling for Efficiency

WaitTimeSeconds=10

Instead of constantly hammering the SQS queue (short polling), the system uses long polling:

Request sent → SQS waits up to 10 seconds for messages to arrive
Messages arrive → Return immediately
No messages → Return empty after 10 seconds, try again

SQS long polling waits for the timeout OR until it hits the MaxNumberOfMessages limit - whichever comes first

This reduces costs and improves responsiveness by minimizing empty API calls.

Potential improvement: During quiet periods (like overnight), the system still polls every 10 seconds. A production system might implement exponential backoff to reduce polling frequency when no messages are available.

3. Message Cleanup and Race Condition Prevention

entries = [{"Id": m["MessageId"], "ReceiptHandle": m["ReceiptHandle"]} for m in messages]
sqs.delete_message_batch(QueueUrl=QUEUE_URL, Entries=entries)

SQS requires both MessageId and ReceiptHandle for deletion. The ReceiptHandle acts like a "secret token" that prevents other consumers or processes from accidentally deleting messages they didn't process.

Message structure:

{
  "MessageId": "12345-abcde-67890",
  "ReceiptHandle": "xyz789...",  // Secret deletion token (changes each time message is received)
  "Body": '{"subject_id": 10000032, "stay_id": 30000123, "intime": "2180-07-23 15:09:00"}',
  "Attributes": {"SentTimestamp": "1625097600000"}
}

Data Processing and Transformation

From JSON Strings to Structured Data

# Convert JSON strings to Spark DataFrame
df = (
    spark.read.json(spark.sparkContext.parallelize(rows), schema=schema)
         .withColumn("intime",  to_timestamp("intime"))
         .withColumn("outtime", to_timestamp("outtime"))
)

The Transformation Pipeline

spark.sparkContext.parallelize(rows): Converts Python list to Spark RDD (Resilient Distributed Dataset) for distributed processing. RDDs are Spark's fundamental data structure - they can be parallelized across multiple CPU cores or machines, unlike regular Python lists.
spark.read.json(..., schema=schema): Parses JSON strings into structured columns with predefined types. It converts the RDD into a Spark DataFrame (higher-level abstraction) using the schema we defined.
to_timestamp(): Converts string timestamps like "2180-07-23 15:09:00" to proper datetime objects for time-based analytics and partitioning.

Partitioning Strategy: The Key to Fast Analytics

The Double Partitioning Approach

df.repartition("intime")          # In-memory organization
  .write
  .partitionBy("intime")          # Disk organization
  .parquet(str(out))

This might look redundant, but each serves a different purpose:

`.repartition("intime")` - Spark Processing Optimization

Purpose: Groups similar timestamps together in Spark's memory partitions before writing to disk.

Example:

Before repartition (data scattered randomly):
Spark Partition 1: [2180-07-23 15:09:00, 2180-07-24 08:30:00]
Spark Partition 2: [2180-07-23 16:15:00, 2180-07-24 14:45:00]

After repartition("intime") (data grouped by time):
Spark Partition 1: [2180-07-23 15:09:00, 2180-07-23 16:15:00]  # Same day grouped
Spark Partition 2: [2180-07-24 08:30:00, 2180-07-24 14:45:00]  # Same day grouped

This prevents creating many small files - without repartitioning, Spark might create multiple files for the same time partition.

`.partitionBy("intime")` - Query Performance Optimization

Purpose: Creates time-based folder structure on disk for fast filtering (called "partition pruning").

Resulting structure:

lake/icu/
├── intime=2180-07-23 15:09:00/
│   └── part-00000.parquet
├── intime=2180-07-23 16:15:00/
│   └── part-00001.parquet
└── intime=2180-07-24 08:30:00/
    └── part-00002.parquet

Why Time-Based Partitioning?

Healthcare analytics are inherently time-focused:

"ICU capacity trends over the past month"
"Average length of stay this quarter vs last quarter"
"Weekend vs weekday admission patterns"

Partition pruning in action:

-- This query only reads July 23rd folders, ignoring everything else
SELECT AVG(los_hrs) 
FROM icu_data 
WHERE intime >= '2180-07-23' AND intime < '2180-07-24'

Performance impact: Instead of scanning millions of records, the query only reads thousands - this is called "partition pruning" where the query engine skips entire partitions based on filter conditions.

Stream Processing Loop: Continuous Operation

The Forever Loop Design

try:
    while True:  # Continuous processing
        rows = next_batch()  # Get messages (blocks until available)

        # Process and transform data
        df = (
            spark.read.json(spark.sparkContext.parallelize(rows), schema=schema)
                 .withColumn("intime", to_timestamp("intime"))
                 .withColumn("outtime", to_timestamp("outtime"))
        )

        # Only save if we have data
        if df.count():
            (df.repartition("intime")
               .write
               .mode("append")
               .partitionBy("intime")
               .parquet(str(out)))

        time.sleep(2)  # Gentle pause between batches

except KeyboardInterrupt:
    print("Stopped.")
    spark.stop()  # Clean shutdown

Why Continuous Processing?

Real-time healthcare monitoring requires:

Immediate processing as patients are admitted/discharged
Always-on availability for 24/7 hospital operations
Graceful handling of quiet periods (no new patients)

Current design considerations:

The 2-second pause prevents overwhelming AWS with requests during processing
The 10-second SQS wait provides natural batching during quiet periods
For production, you'd want exponential backoff during extended quiet periods

Error Handling and Graceful Shutdown

Keyboard interrupt handling ensures:

Current batch completes processing (data consistency)
Spark resources are properly released (memory cleanup)
No data loss from incomplete operations
Clean connection termination to AWS services

Development Strategy: Local First, Cloud Later

Configuration Flexibility

OUTPUT_DIR = os.environ.get("VW_LAKE_DIR", "./lake/icu")  # Default: local

Development workflow:

# Start with local testing (Spark runs locally on your machine)
python stream_sqs_to_parquet.py

# Scale to S3 when ready (same code, different output location)
export VW_LAKE_DIR="s3://your-bucket/data/"
python stream_sqs_to_parquet.py

Benefits of local-first approach:

Faster iteration (no network delays for file I/O)
Cost control (no AWS storage charges during development)
Easy debugging (can inspect Parquet files directly with tools like parquet-tools)
Same code works for both local and cloud storage (abstraction layer)

Real-World Impact: From Raw Messages to Healthcare Insights

Before: Raw SQS Messages

{"subject_id": 10000032, "stay_id": 30000123, "intime": "2180-07-23 15:09:00", "outtime": "2180-07-31 13:58:00", "los_hrs": 190.82}

After: Analytics-Ready Data Lake

lake/icu/  (or s3://vitalwatch-lake/)
├── intime=2180-07-23 15:09:00/
│   ├── part-00000.parquet (compressed, columnar, schema-enforced)
│   └── _SUCCESS (Spark success marker)
├── intime=2180-07-23 16:15:00/
│   ├── part-00001.parquet
│   └── _SUCCESS

Future Analytics Possibilities

-- Hospital capacity planning
SELECT DATE(intime), COUNT(*) as daily_admissions
FROM icu_data 
WHERE intime >= '2024-01-01'
GROUP BY DATE(intime);

-- Length of stay analysis
SELECT AVG(los_hrs) as avg_los
FROM icu_data 
WHERE intime >= current_date() - interval 30 days;

-- Peak hours identification  
SELECT HOUR(intime), COUNT(*) as admissions
FROM icu_data 
GROUP BY HOUR(intime)
ORDER BY admissions DESC;

Key Learnings and Insights

1. Stream Processing vs Batch Processing

Traditional ETL processes data in large chunks once per day. Stream processing handles data as it arrives, enabling:

Real-time dashboards for hospital administrators
Immediate alerts for unusual patterns
Up-to-date analytics for clinical decision-making

Trade-off: Stream processing adds complexity but enables real-time insights.

2. The Power of Columnar Storage

Switching from row-based (CSV) to column-based (Parquet) storage isn't just a technical detail:

Query performance improves by 10-100x for analytics workloads
Storage costs decrease by 50-80% due to column-specific compression
Data processing scales to billions of records efficiently

3. Partition Strategy Matters

Time-based partitioning aligns with how healthcare data is actually queried:

Most questions involve time ranges ("last month", "this quarter")
Regulatory reporting is time-period based
Operational metrics track daily/weekly/monthly trends

4. Graceful Degradation and Production Considerations

Current system handles basic scenarios but production systems need:

Exponential backoff during quiet periods
Dead letter queues for message processing failures
Monitoring and alerting for system health
Batch size optimization for throughput

Current Limitations and Future Improvements

What Could Be Enhanced:

True batch accumulation: Currently limited to 10 messages per batch due to SQS API limits
Quiet period optimization: System polls every 10 seconds even when no data is flowing
Error handling: Basic error handling - production would need retry logic, dead letter queues
Monitoring: No metrics on processing latency, throughput, or system health

For Production Deployment:

Spark cluster instead of local Spark instance
Auto-scaling based on queue depth
Multiple consumer instances for higher throughput
Data quality monitoring and alerting

What's Next: Phase 3

With Phase 2 complete, we now have:

✅ Real-time data ingestion (Phase 1)
✅ Stream processing pipeline (Phase 2)
✅ Analytics-ready data lake with columnar storage and time-based partitioning

Phase 3 will add:

DynamoDB for real-time alerts storage
FastAPI web service for dashboard APIs
WebSocket connections for live updates to browser clients
Kubernetes deployment for production scaling and reliability

Resources and Code

The complete Phase 2 implementation is available in my VitalWatch repository.

Key files:

stream_sqs_to_parquet.py - Main stream processing script
pyproject.toml - Project dependencies and configuration
cleanup.sh - Resource cleanup for cost control

Technical Specifications

Dependencies:

pip install pyarrow==16.1 awscli boto3

We use pyarrow because it allows faster read/write for Spark to process Parquet files

AWS Services Used:

SQS: Message queuing and stream buffering (FIFO queues for ordered processing)
S3: Data lake storage for Parquet files (optional, can start local)
IAM: Access control and permissions

Local Requirements:

Apache Spark 3.5: Stream processing engine (runs locally for development)
Python 3.8+: Runtime environment

This is Part 2 of the VitalWatch series. Next up: "Phase 3: Building Real-time Healthcare Dashboards with FastAPI and WebSockets"

Series Navigation:

Part 1: Building the Data Ingestion Pipeline
Part 2: Stream Processing with Apache Spark ← You are here
Part 3: Real-time Dashboards and Alerts ← Coming soon

Building My First Real-Time Healthcare Data Pipeline with AWS SQS and Python

jayanti-neu — Fri, 18 Jul 2025 13:52:49 +0000

Building a Real-Time Healthcare Data Pipeline: From MIMIC-IV to AWS SQS

Today I learned how to build the foundation of a real-time healthcare monitoring system using AWS SQS and Python. This is Phase 1 of a larger project called "VitalWatch" that will eventually process ICU patient data in real-time to detect anomalies and send alerts.

The Big Picture: What We're Building

VitalWatch is a healthcare monitoring system with three phases:

Phase 1: Stream vital signs from MIMIC-IV dataset to AWS SQS (what I built today)
Phase 2: Process the stream with Spark to calculate rolling averages and detect anomalies
Phase 3: Build a web dashboard with real-time alerts

Think of it like the monitoring systems in hospital ICUs - continuous streams of patient data that need to be processed instantly to catch medical emergencies.

Understanding the Data: MIMIC-IV

The MIMIC-IV database contains real anonymized ICU data from Beth Israel Deaconess Medical Center. The full dataset has:

364,627 unique patients
546,028 hospitalizations
94,458 ICU stays
Hundreds of millions of vital sign measurements

For learning purposes, I used the demo dataset (100 patients) to avoid AWS charges while getting the same data structure.

Why AWS SQS? And Why FIFO?

I considered several alternatives for streaming healthcare data:

SQS vs Real Alternatives

WebSockets: Great for real-time browser communication, but wrong for this use case

❌ No persistence (messages lost if connection drops)
❌ No reliability guarantees
❌ Requires both ends to be online simultaneously
✅ Good for: Live chat, real-time dashboards

RabbitMQ: Powerful message broker, but more complex

✅ More routing options and features
❌ You have to install and manage it yourself
❌ Need to handle clustering, persistence, monitoring
✅ Good for: Complex enterprise messaging patterns

Apache Kafka: Built for high-throughput streaming

✅ Handles massive data volumes
✅ Messages persist for replay
❌ Much more complex to set up and operate
❌ Overkill for this project size
✅ Good for: Large-scale event streaming, data lakes

SQS advantages for my use case:

Reliability: Messages are guaranteed to be delivered
Persistence: Messages wait in queue until processed
Decoupling: Producer and consumer don't need to be online simultaneously
Scaling: AWS handles the infrastructure automatically
Simplicity: No servers to manage, just send/receive messages

Why FIFO over Standard SQS:
For medical data, order matters. Consider these scenarios:

Blood pressure: 120 → 80 → 60 (concerning downward trend)
vs: 60 → 80 → 120 (patient recovering)

Processing these out of order could lead to wrong clinical conclusions.

Setting Up the AWS Infrastructure

Step 1: Create the FIFO Queue

# Create a FIFO queue (free: 1M requests/month)
QUEUE_NAME="vitalwatch-demo.fifo"
aws sqs create-queue \
  --queue-name "$QUEUE_NAME" \
  --attributes FifoQueue=true,ContentBasedDeduplication=true

Key attributes explained:

FifoQueue=true: Ensures messages are processed in exact order
ContentBasedDeduplication=true: Prevents duplicate vital signs if the producer accidentally sends the same reading twice

Step 2: Save Queue URL for Later Use

export VW_QUEUE_URL=$(aws sqs get-queue-url --queue-name "$QUEUE_NAME" --query 'QueueUrl' --output text)
echo "Queue URL => $VW_QUEUE_URL"

This command does several things:

aws sqs get-queue-url - Asks AWS for the queue's full URL
--query 'QueueUrl' - Extracts just the URL from the JSON response
--output text - Removes quotation marks around the URL
export - Makes the URL available only in this terminal session

Important: The export command only saves the variable in your current terminal session. If you:

Close the terminal
Open a new terminal tab/window
Log out and back in

You'll need to run the export command again. The variable isn't saved permanently to your system - it's just stored in your current shell's memory until that session ends.

Building the Data Producer

Python Project Structure

I learned about modern Python project organization:

vitalwatch/
├── pyproject.toml      # Modern Python project config
├── cleanup.sh          # Resource cleanup script
└── src/
    └── vitalwatch/
        ├── __init__.py
        └── producer.py

Modern Python Configuration

Instead of the old setup.py, I used pyproject.toml:

[project]
name = "vitalwatch"
version = "0.1.0"

This is Python's equivalent to package.json in Node.js - it defines project metadata and dependencies.

Installing in Development Mode

pip install -e .

The -e flag creates a link to your source code instead of copying it. Let me explain the difference:

Without -e (normal install):

pip install .

Copies your code to the virtual environment's site-packages folder
If you edit producer.py, the changes won't be reflected until you reinstall
Like installing a finished app from an app store

With -e (editable/development install):

pip install -e .

Creates a link pointing to your source code location
Changes to your code are immediately available
No need to reinstall after each edit
Perfect for active development

Real example:

# After pip install -e .
python -m vitalwatch.producer  # Runs your current code

# Edit producer.py and save changes
# Now run again:
python -m vitalwatch.producer  # Uses your updated code immediately!

Think of it like creating a shortcut on your desktop vs. copying the entire program folder.

The Producer Script

import gzip, csv, json, os, sys, time, uuid
from datetime import datetime, timezone
import boto3

# Configuration
DEMO_CSV = os.path.expanduser(
    "~/mimic-demo/mimic-iv-clinical-database-demo-2.2/icu/icustays.csv.gz"
)
QUEUE_URL = os.environ.get("VW_QUEUE_URL") or sys.exit("VW_QUEUE_URL not set")

# Add throttling control
ROWS_PER_SEC = float(os.environ.get("ROWS_PER_SEC", "0.2"))  # Default to 0.2
SLEEP_TIME = 1.0 / ROWS_PER_SEC

sqs = boto3.client("sqs")

def iter_rows(path: str):
    with gzip.open(path, "rt", newline="") as fh:
        reader = csv.DictReader(fh)
        for row in reader:
            yield {
                "subject_id": int(row["subject_id"]),
                "stay_id": int(row["stay_id"]),
                "intime": row["intime"],
                "outtime": row["outtime"],
                "los_hrs": float(row["los"]),
            }

def main():
    for row in iter_rows(DEMO_CSV):
        sqs.send_message(
            QueueUrl=QUEUE_URL,
            MessageBody=json.dumps(row),
            MessageGroupId="icustays",
            MessageDeduplicationId=str(uuid.uuid4()),
        )
        print("sent", row)
        time.sleep(SLEEP_TIME)

Key Learning:

The `yield` Keyword

The yield keyword creates a generator - it's like a pause button for functions:

Without yield (memory intensive):

def get_all_rows():
    rows = []
    for line in file:
        rows.append(process_line(line))  # Loads ALL rows into memory
    return rows

With yield (memory efficient):

def iter_rows():
    for line in file:
        yield process_line(line)  # Processes one row at a time

For streaming data, this is crucial - you process one piece at a time without overwhelming your system's memory.

SQS Send Message

sqs.send_message(
            QueueUrl=QUEUE_URL,
            MessageBody=json.dumps(row),
            MessageGroupId="icustays",
            MessageDeduplicationId=str(uuid.uuid4()),
        )

What Happens After This Call:

AWS receives the message and stores it in the FIFO queue
It maintains order within the "icustays" group and waits for a consumer to read it while also preventing duplicates using the deduplication ID

This message sits in the queue until something (like the smoke test command - scroll down to see that) reads it!

Dependencies Explained

pip install boto3 pandas python-dateutil

boto3: AWS SDK for Python - lets your code communicate with AWS services
pandas: Data manipulation library (will be used more in Phase 2 with Spark)
python-dateutil: Robust date/time parsing for medical timestamps

Staying Within AWS Free Tier

The free tier allows 1M SQS requests per month. At 1 message/second, that's 2.6M messages/month if running 24/7.

Cost Control Strategy

I added throttling to stay within limits:

# Add to producer script
ROWS_PER_SEC = float(os.environ.get("ROWS_PER_SEC", "0.2")) 
SLEEP_TIME = 1.0 / ROWS_PER_SEC
time.sleep(SLEEP_TIME)

Usage:

# Safe for free tier (default)
python -m vitalwatch.producer

# For demos (watch your usage)
export ROWS_PER_SEC=1.0
python -m vitalwatch.producer

Monitoring and Cleanup

Created a cleanup script to delete all resources:

#!/bin/bash
# cleanup.sh
aws sqs delete-queue --queue-url "$VW_QUEUE_URL"
# Plus cleanup for future phases (S3, DynamoDB, etc.)

Testing the Pipeline (SMOKE TEST)

Start the Producer

python -m vitalwatch.producer

In Another Terminal, Test Message Receiving

aws sqs receive-message \
  --queue-url "$VW_QUEUE_URL" \
  --max-number-of-messages 5 \
  --visibility-timeout 0

This shows you the messages in JSON format, proving the pipeline works.

Key Insights and Learnings

1. Order Matters in Healthcare Data

FIFO queues are essential for medical data where the sequence of events determines patient outcomes.

2. Modern Python Project Structure

Using pyproject.toml and pip install -e . creates a professional, maintainable project structure.

3. Memory-Efficient Data Processing

Using yield allows processing large datasets without overwhelming system memory.

4. Cloud vs Local Trade-offs

While this requires internet connectivity, the reliability and scalability benefits of cloud infrastructure outweigh the complexity for production systems.

5. Cost Awareness

Even "free" cloud services have limits. Building in throttling and monitoring from the start prevents unexpected charges.

What's Next?

Phase 2 will involve:

Setting up Spark Structured Streaming
Reading from the SQS queue in real-time
Computing rolling averages and anomaly detection
Writing results to S3 in Parquet format

This foundation makes all of that possible - the producer will keep streaming data while Spark processes it continuously.

Final Thoughts

Building this taught me that real-time data pipelines aren't just about the technology - they're about understanding the domain (healthcare), the constraints (free tier limits), and the trade-offs (reliability vs complexity).

The most important lesson: Start simple, but start with the right architecture. This basic producer-queue-consumer pattern scales from demo datasets to production systems handling millions of messages.

Resources

This is Part 1 of a 3-part series on building VitalWatch, a real-time healthcare monitoring system. Stay tuned for Part 2: "Real-time Analytics with Spark Structured Streaming"

🚦 Real-Time Freight Tracker with Java & WebSockets (Phase 2)

jayanti-neu — Sat, 12 Jul 2025 17:11:32 +0000

After building the core CRUD backend in Phase 1, Phase 2 of the Freight Tracker project adds real-time updates using WebSockets. This allows clients to receive live status changes for shipments without needing to refresh the page.

Its continuation of building a full-stack logistics tracking app, this post adds real-time updates using Spring Boot, STOMP, and WebSockets.

🌐 Why Real-Time with WebSockets?

In logistics applications, users (like clients, warehouse staff, or support teams) often need instant updates on shipment status. Polling the backend repeatedly isn’t efficient — WebSockets provide a two-way communication channel where the server pushes updates to connected clients instantly.

Spring Boot supports this via:

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

⚙️ Setting up WebSockets in Spring Boot

WebSocketConfig.java

We created a new class called WebSocketConfig.java to enable STOMP messaging and define endpoints:

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
                .setAllowedOriginPatterns("*")
                .withSockJS();
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic");
        registry.setApplicationDestinationPrefixes("/app");
    }
}

Two Sides of WebSocket Communication

When using STOMP over WebSockets in Spring, there are two types of message flow:

Client → Server (incoming messages)

Server → Client (broadcast messages)

We configure these in two parts:

registerStompEndpoints(): Where clients initially connect (like /ws)

configureMessageBroker(): How messages are routed internally (topics, queues, etc.)

🔑 Keywords Explanation:

@Configuration: Marks the class as a Spring config class which means it processes the class during startup
@EnableWebSocketMessageBroker: Turns on WebSocket messaging with STOMP protocol and enables the broker system (/topic or /queue). It allows you to also map /app/... for client → server messages
.addEndpoint("/ws"): Clients connect to this endpoint
withSockJS(): Adds fallback support if browser doesn't support native WebSockets
enableSimpleBroker("/topic"): Clients subscribe to these topics to receive updates
setApplicationDestinationPrefixes("/app"): Prefix for client-to-server messages

📢 Broadcasting Status Updates

To send messages to subscribed clients, we created a broadcaster class:

ShipmentStatusBroadcaster.java

@Component
public class ShipmentStatusBroadcaster {
    @Autowired
    private SimpMessagingTemplate messagingTemplate;

    public void broadcastUpdate(ShipmentUpdateMessage message) {
        messagingTemplate.convertAndSend("/topic/shipments", message);
    }
}

SimpMessagingTemplate is used to send messages to WebSocket subscribers (like RestTemplate is used for HTTP).

Why Do We Need SimpMessagingTemplate?
The service has business events (e.g., shipment status updated) and we want to push those events to clients manually — not in response to a client message, but whenever the backend decides (like after saving to DB).

SimpMessagingTemplate is a helper that talks directly to the broker we created.

SimpMessagingTemplate.convertAndSend() = Pushes events into the broker

🧠 What is `@Component`?

This marks the class as a Spring-managed bean. It gets picked up automatically during component scanning and injected wherever needed (like in the ShipmentController).

📦 Updating the Controller

We modified the ShipmentController to send broadcast messages on every shipment creation or update:

@PostMapping
public Shipment createShipment(@RequestBody Shipment shipment) {
    shipment.setLastUpdatedTime(LocalDateTime.now());
    Shipment saved = shipmentRepository.save(shipment);

    broadcaster.broadcastUpdate(
        ShipmentUpdateMessage.builder()
            .shipmentId(saved.getId())
            .trackingNumber(saved.getTrackingNumber())
            .status(saved.getStatus())
            .lastUpdatedTime(saved.getLastUpdatedTime().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME))
            .build()
    );

    return saved;
}

We do the same in the PUT /api/shipments/{id} method.

Frontend (subscribe) ----connect---> /ws endpoint
|
| Subscribes to /topic/shipments
v
Spring Broker (enabled by enableSimpleBroker)
|
| convertAndSend("/topic/shipments", dto)
v
Frontend receives update in real-time

📨 DTO for Real-Time Updates

ShipmentUpdateMessage.java

@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class ShipmentUpdateMessage {
    private Long shipmentId;
    private String trackingNumber;
    private ShipmentStatus status;
    private String lastUpdatedTime;
}

We keep this DTO separate from the main Shipment model to ensure:

Lightweight messages
No overexposing of sensitive data
Easy control over WebSocket response structure

🔌 Frontend Client Example

Here’s how a basic HTML frontend listens for shipment updates:

<script src="https://cdn.jsdelivr.net/npm/sockjs-client@1/dist/sockjs.min.js"></script>
<script src="https://cdn.jsdelivr.net/npm/stompjs@2.3.3/lib/stomp.min.js"></script>

<script>
const socket = new SockJS("http://localhost:8080/ws");
const stompClient = Stomp.over(socket);

stompClient.connect({}, function (frame) {
  console.log("Connected: " + frame);
  stompClient.subscribe("/topic/shipments", function (message) {
    const body = JSON.parse(message.body);
    console.log(`Shipment #${body.trackingNumber} is now ${body.status} at ${body.lastUpdatedTime}`);
  });
});
</script>

Every time a shipment is created or updated, this client will receive a message in real time.

🧠 What I Learned in Phase 2

How STOMP protocol simplifies WebSocket communication
How to use @EnableWebSocketMessageBroker to configure endpoints
How SimpMessagingTemplate is used to push messages
When and why to use DTOs for WebSockets
How real-time updates improve client experience in logistics

🗓️ Coming Next

React-based frontend dashboard
Service Layer refactor and error handling
Deployment to cloud with Docker and AWS

Thanks for reading — you can find the code here: https://github.com/jayanti-neu/freight-tracker 🚀

Building a Real-Time Freight Tracker in Java — Beginner’s Perspective

jayanti-neu — Thu, 10 Jul 2025 22:35:17 +0000

🚚 Building a Real-Time Freight Tracker in Java (Phase 1)

As a recent graduate stepping into backend development with Java Spring Boot, I wanted to build something real-world and practical. I've always been curious about how delivery tracking systems work so I started working on a Real-Time Freight Tracking Application.

This blog is a beginner-friendly, exploratory recap of how I built Phase 1: the core CRUD API backend.

🌟 Project Goal

Build a backend system that allows tracking of freight shipments across different locations with key details like origin, destination, status, and timestamps and more

🔧 Tech Stack Used

Java 17
Spring Boot 3.5
Maven
PostgreSQL (local DB)
IntelliJ IDEA (Community Edition)
Postman (for testing)

🧠 How the Code Works — MVC and Data Flow

🧹 Model Layer: The Domain

The core data structure is Shipment — a JPA entity mapped to a database table. Here's what it looks like:

@Entity
public class Shipment {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String origin;

    private String destination;

    @Enumerated(EnumType.STRING)
    private ShipmentStatus status;

    private LocalDateTime lastUpdatedTime;

    @Column(nullable = false, unique = true)
    private String trackingNumber;

    private String carrier;

    @Enumerated(EnumType.STRING)
    private Priority priority;

}

What I learned here:

@Entity marks it as a database-mapped object
@Id and @GeneratedValue handle primary key logic
@Enumerated(EnumType.STRING) saves enums as readable strings

The ShipmentStatus enum represents fixed states:

public enum ShipmentStatus {
    PENDING, IN_TRANSIT, DELIVERED, CANCELLED
}

And Priority (another enum) tracks urgency levels.

Lombok annotations like @Getter, @Setter, @Builder, etc. reduce boilerplate code.

🗂️ Repository Layer: Data Access

This layer connects our code with the database:

@Repository
public interface ShipmentRepository extends JpaRepository<Shipment, Long> {
    Page<Shipment> findByStatus(ShipmentStatus status, Pageable pageable);
    Page<Shipment> findByOriginIgnoreCase(String origin, Pageable pageable);
    Page<Shipment> findByOriginIgnoreCaseAndStatus(String origin, ShipmentStatus status, Pageable pageable);
    Page<Shipment> findAll(Pageable pageable);
    long countByStatus(ShipmentStatus status);

    @Query("SELECT s.origin FROM Shipment s GROUP BY s.origin ORDER BY COUNT(s) DESC LIMIT 1")
    String findMostCommonOrigin();
}

Key Concepts:

Spring Data JPA uses method names to auto-generate SQL
Pagination is supported using Pageable — allowing APIs like: GET /api/shipments?page=0&size=10 or GET /api/shipments/search?origin=NY&status=IN_TRANSIT&page=1&size=5
The custom @Query uses JPQL to find the most common origin efficiently

This structure ensures scalability and efficiency.

🌐 Controller Layer: API Endpoints

Here's a simplified controller class:

@RestController
@RequestMapping("/api/shipments")
public class ShipmentController {

    @Autowired
    private ShipmentRepository shipmentRepository;

    @PostMapping
    public Shipment createShipment(@RequestBody Shipment shipment) {
        shipment.setLastUpdatedTime(LocalDateTime.now());
        return shipmentRepository.save(shipment);
    }

    @GetMapping
    public List<Shipment> getAllShipments() {
        return shipmentRepository.findAll();
    }

    @GetMapping("/search")
    public List<Shipment> searchShipments(
        @RequestParam(required = false) String origin,
        @RequestParam(required = false) ShipmentStatus status,
        Pageable pageable
    ) {
        if (origin != null && status != null) {
            return shipmentRepository.findByOriginIgnoreCaseAndStatus(origin, status, pageable);
        } else if (origin != null) {
            return shipmentRepository.findByOriginIgnoreCase(origin, pageable);
        } else if (status != null) {
            return shipmentRepository.findByStatus(status, pageable);
        } else {
            return shipmentRepository.findAll(pageable);
        }
    }
    // more endpoints...
}

Controller Annotations Explained:

@RestController: Shortcut for @Controller + @ResponseBody, It automatically converts your return values (like Java objects) into JSON responses
@RequestMapping: Base route for the class
@Autowired: Dependency injection of the repository
@PostMapping, @GetMapping: Maps HTTP methods to Java methods
@RequestParam: Handles query strings like ?origin=NY
@RequestBody: Converts incoming JSON into Java objects

This is where HTTP meets business logic.

📊 Stats & DTOs

To keep response structure clean and decoupled from the database model, I used a DTO:

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class ShipmentStatsDTO {
    private long totalShipments;
    private Map<ShipmentStatus, Long> statusCounts;
    private String mostCommonOrigin;
}

Why DTOs?

Keeps internal model logic separate from external API responses — this is essential for security and flexibility
Prevents overexposing database fields, such as sensitive IDs or internal flags
Makes testing and documentation easier, since DTOs define exactly what data is expected or returned
Helps decouple frontend and backend development — frontend teams can rely on stable response shapes even if internals change
Promotes consistency in API design by enforcing contract-based responses
Improves serialization performance and clarity by shaping only the needed data (e.g., filtering fields, renaming keys)
Makes integration with tools like Swagger/OpenAPI or frontend types (like TypeScript interfaces) cleaner and easier to auto-generate
Aids in unit testing: you can test service logic using DTOs directly without loading full entities
Especially useful for read-heavy APIs or summary/analytics endpoints where the result doesn't mirror a database table model logic separate from API responses
Prevents overexposing database fields
Simplifies and formats responses (like maps and summaries)

The /api/shipments/stats endpoint returns this structure.

🔧 Configuration & Secrets Management

To avoid committing passwords:

Gitignored application.properties
Used application.properties.example for public template

🧳 Directory Structure Summary

freight-tracker/
├── model/                # Java classes like Shipment, ShipmentStatus, Priority
├── repository/           # Interfaces for DB queries (Spring Data JPA)
├── controller/           # RESTful HTTP endpoints
├── dto/                  # DTOs for stats and summaries
├── FreightTrackerApplication.java  # Main entry point
├── application.properties.local

🧠 What I Learned in Phase 1

How MVC architecture connects models, repos, and controllers
How to build clean REST APIs using Spring Boot annotations
How enums and timestamps are stored using JPA
Query filtering using @RequestParam
Creating DTOs to shape and return custom responses
Adding pagination to repository methods for real-world scalability
Structuring code professionally (modularity, separation of concerns)
Secrets management using .gitignore
Endpoints tested with Postman

🗓️ What’s Coming Next

Real-time updates with WebSockets
Service layer refactor
UI or dashboard (React or Thymeleaf)
AWS RDS and cloud deployment

👩‍💻 Final Thoughts

You can check out the code on GitHub [https://github.com/jayanti-neu/freight-tracker], and stay tuned for more updates!

If you have questions or feedback, feel free to connect with me. 🚀

DEV Community: jayanti-neu

🚛 Freight Tracker: What I Did After Polishing the UI

🔄 Making the App Feel More Dynamic with Zustand and WebSockets

🗺️ Making the Map More Realistic

⚠️ Fixing the Refresh Issue on Netlify

🕒 Preventing Backend Sleep with GitHub Actions

🚛 Freight Tracker Frontend — Phase 2 & Phase 3 Recap

🧭 Overview

📦 Phase 2: CRUD Pages, State Management & Styling

✅ 1. Shipment Type Definitions

🧠 2. Global State with Zustand

🖼️ 3. Basic Pages with Routing

🖍️ 4. Styling with TailwindCSS

⚡ Phase 3: WebSocket Updates, Maps, and Visualization

🌐 1. Real-Time WebSocket Integration

🗺️ 2. Interactive Map with Leaflet

💡 3. State-Driven Map Updates

🚦 4. Small Optimizations

🧠 Lessons Learned

📌 Next Steps

🚀 Building the Freight Tracker Frontend (Phase 1)

🎯 Goals of Phase 1

🛠 Tech Stack

1. Project Setup

Create Vite project

Install Tailwind (latest)

2. Professional Folder Structure

Why this structure?

3. What is an SPA and Why React Router?

How SPAs Work

Why React Router DOM?

4. Implementing Routing

App.tsx

pages/HomePage.tsx

pages/ShipmentsPage.tsx

How Navigation Works

5. Key Learnings from Phase 1

6. What's Next?

GitHub Repo

🚚 Building a Real-Time Freight Tracker in Java (Phase 3)

🧠 Why Refactor?

1️⃣ Service Layer

ShipmentService.java

ShipmentServiceImpl.java

2️⃣ Centralized Error Handling

3️⃣ DTOs + @valid

DTO (Data Transfer Object)

@valid

Difference Between DTO and @valid

4️⃣ Testing

Unit Test Example (Service Layer)

Integration Test Example (Controller + H2 DB)

Why both Unit and Integration?

🗓️ What's Next?

👩‍💻 Final Thoughts

Building a Real-Time Healthcare Data Pipeline with Apache Spark: From SQS to Parquet (Part 2)

Building a Real-Time Healthcare Data Pipeline - Phase 2: Stream Processing with Apache Spark

The Evolution: From Message Queue to Data Lake

Where We Left Off

The Big Picture: What Phase 2 Accomplishes

Understanding the Technology Stack - Key Technologies Explained

Why Apache Spark for Stream Processing?

The Storage Revolution: Parquet Format

Architecture Deep Dive

Data Schema Definition

The Stream Processing Pipeline

Key Design Decisions Explained

1. Batch Processing with SQS Limits

2. Long Polling for Efficiency

3. Message Cleanup and Race Condition Prevention

Data Processing and Transformation

From JSON Strings to Structured Data

The Transformation Pipeline

Partitioning Strategy: The Key to Fast Analytics

The Double Partitioning Approach

.repartition("intime") - Spark Processing Optimization

.partitionBy("intime") - Query Performance Optimization

Why Time-Based Partitioning?

Stream Processing Loop: Continuous Operation

The Forever Loop Design

`.repartition("intime")` - Spark Processing Optimization

`.partitionBy("intime")` - Query Performance Optimization

The `yield` Keyword

🧠 What is `@Component`?