Imagine you're a security guard at a high-end jewelry store. You aren't just watching the front door; you're watching a continuous stream of people walking in and out. Now, a single person entering the store is normal. But if you see the same person enter the store, leave, and come back five times in ten minutes—each time wearing a different disguise—you're going to step in. You aren't judging them based on a single snapshot in time, but on a pattern of behavior over a specific window of time.

That is exactly how we build a fraud detection pipeline in Spark. We aren't looking for a single "bad" transaction; we're looking for "bad" sequences. To do this, we map our security guard's logic directly into Spark Structured Streaming:

• The Flow of People: This is our DataStreamReader, pulling transactions from a source like Kafka.
• The Disguises/IDs: These are our grouping keys (like userId or creditCardNumber).
• The Ten-Minute Window: This is the window() function, which tells Spark to bucket events into time-slices.
• The Red Flag: This is our filter() condition—for example, "more than 3 transactions in 10 minutes from different IP addresses."

Defining the Red Flags in Code

When I first started writing these pipelines, I tried to handle the state manually using a database. It was a nightmare of race conditions. With Spark, we use groupBy combined with a window. Let's look at how we catch a "velocity attack"—where a card is swiped too many times in a short burst.

import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.Trigger

case class Transaction(userId: String, amount: Double, timestamp: java.sql.Timestamp, location: String)

val transactions = spark.readStream
  .schema(transactionSchema)
  .json("/data/transactions")

val fraudAlerts = transactions
  .withWatermark("timestamp", "10 minutes") // Handle late-arriving data
  .groupBy(
    window($"timestamp", "10 minutes", "5 minutes"), 
    $"userId"
  )
  .agg(count("*").as("tx_count"), collect_set($"location").as("distinct_locations"))
  .filter($"tx_count" > 3 && size($"distinct_locations") > 1)

fraudAlerts.writeStream
  .outputMode("update")
  .format("console")
  .start()

Dealing with the "Late Arrival" Problem

In the real world, data is messy. A mobile app might lose signal and upload a transaction five minutes after it actually happened. If your window closes at 10:00, and a 09:58 transaction arrives at 10:02, you don't want to ignore it, but you also can't keep the window open forever or you'll run out of memory.

That's why I included withWatermark in the code above. Think of the watermark as a "grace period." By setting it to 10 minutes, you're telling Spark: "Wait for late data, but once the system clock hits 10:15, I no longer care about anything that happened before 10:05." It's a trade-off between perfect accuracy and system stability. I've seen clusters crash because engineers forgot watermarks and the state grew until the JVM exploded.

Tuning the Sliding Window

You'll notice I used window($"timestamp", "10 minutes", "5 minutes"). The first duration is the window length, and the second is the slide duration.

If the slide is the same as the length, you have "tumbling windows" (they don't overlap). But for fraud, we usually want "sliding windows." By sliding every 5 minutes, we ensure that a burst of fraud spanning the boundary of two tumbling windows doesn't slip through the cracks. It's like the security guard checking the logs every 5 minutes rather than once every 10.

📋 Practical Task

Build a High-Velocity Geographic Jump Detector

Your task is to extend the fraud pipeline to detect "Impossible Travel." This occurs when a user makes transactions in two different cities within a time window that would be physically impossible to travel between.

Requirements:

• Create a Spark Structured Streaming pipeline that processes a stream of Transaction objects (userId, amount, timestamp, city).
• Implement a sliding window of 30 minutes, sliding every 10 minutes.
• Use collect_set("city") to track all unique cities a user visited within that window.
• Filter for users who have appeared in more than 2 different cities within that 30-minute window.
• Apply a watermark of 15 minutes to handle delayed events.
• Output the results to the console in update mode, showing the userId and the list of cities they "visited."