Furkan İkkan Türkçe
Back to posts
Post

· Network & System Administrator

How to Prevent Data Loss in Spark Structured Streaming with Watermark and allowedLateness

Learn how to use watermark and allowedLateness in Spark Structured Streaming to handle late events and avoid data loss in event-time processing.

Big DataSpark Structured StreamingwatermarkallowedLatenessevent-time processingdata loss prevention

When I first started working with Spark Structured Streaming for real-time analytics, I assumed that as long as my job was running, all events would be processed correctly. But I quickly learned that event-time processing introduces a silent risk: late-arriving data. If your stream doesn’t account for delays — whether from network issues, batch ingestion lag, or mobile devices syncing offline — you can lose valuable events without any obvious error. In this post, I’ll share how I use watermark and allowedLateness to prevent data loss in my production pipelines, based on real incidents I’ve debugged.

Understanding Event-Time Processing and the Problem of Late Data

In event-time processing, Spark doesn’t process data based on when it arrives (processing time), but on the timestamp embedded in the event itself. This is crucial for use cases like user sessionization, fraud detection, or IoT telemetry where the timing of the event matters more than when we saw it.

However, real-world systems are imperfect. Events can arrive minutes, hours, or even days late. If your streaming job has already advanced its event-time watermark past the timestamp of a late event, Spark will drop it silently — no warning, no metric spike, just missing data in your downstream aggregates.

I’ve seen this happen in a retail analytics pipeline where mobile app events from users in areas with poor connectivity were delayed by up to 4 hours. Our initial watermark setting of 10 minutes meant we were losing nearly 15% of evening sales events. That’s not acceptable when revenue reporting depends on accuracy.

How Watermark Works in Spark Structured Streaming

The watermark is Spark’s mechanism for tracking how far behind event time we’re willing to wait for late data. It’s defined as:

watermark = max_event_time_seen - delay_threshold

Where delay_threshold is the maximum lateness you expect. Any event with a timestamp older than the current watermark is considered too late and is dropped.

You set it in your query like this:

val stream = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
  .option("subscribe", "user-events")
  .load()
  .selectExpr("CAST(value AS STRING)")
  .select(from_json(col("value"), schema).as("data"))
  .select("data.userId", "data.eventTime", "data.action")
  .withWatermark("eventTime", "2 hours")

val aggregated = stream
  .groupBy(window(col("eventTime", "1 hour"), "15 minutes"), col("userId"))
  .agg(count("*").as("eventCount"))

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

In this example, we’re telling Spark: "Keep state for events up to 2 hours behind the latest event time you’ve seen." If an event arrives with an eventTime of 10:00 AM, and the latest event we’ve seen is at 2:00 PM, the watermark is 12:00 PM — so the 10:00 AM event is still processed. But if it arrives after 2:00 PM, it’s dropped.

The Role of allowedLateness: Giving Late Data a Second Chance

Watermark alone is strict: once passed, data is gone. But sometimes, you know lateness can exceed your watermark threshold due to rare but predictable delays — like daily batch uploads from partner systems that run at 3 AM.

That’s where allowedLateness comes in. It extends the window beyond the watermark for stateful operations like window() or dropDuplicates(). Events arriving after the watermark but within the allowed lateness period can still update state.

Here’s how I use it in a payment reconciliation stream:

val payments = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "payments-broker:9092")
  .option("subscribe", "payment-events")
  .load()
  .selectExpr("CAST(value AS STRING)")
  .select(from_json(col("value"), paymentSchema).as("data"))
  .select("data.transactionId", "data.eventTime", "data.amount", "data.status")
  .withWatermark("eventTime", "30 minutes")

val dailySummary = payments
  .groupBy(window(col("eventTime", "1 day"), "1 hour"), col("status"))
  .agg(sum("amount").as("totalAmount"))
  .allowLateness("6 hours")  // Allow updates up to 6 hours late

val query = dailySummary
  .writeStream
  .outputMode("update")
  .format("delta")
  .option("checkpointLocation", "/mnt/delta/checkpoints/payments_daily")
  .trigger(Trigger.ProcessingTime("10 minutes"))
  .start()

In this case, even if a payment event from yesterday arrives at 9 AM today (15 hours late), as long as it’s within the 6-hour allowed lateness window relative to the window’s end time, it can still update the daily total. Without allowLateness("6 hours"), those events would be ignored.

Best Practices I Follow in Production

Over time, I’ve developed a few rules that help me avoid data loss while keeping state size manageable:

  • Start with observed latency: Measure your actual event delay in staging. Use the difference between eventTime and ingestionTime (when Kafka received it) to set a realistic watermark.
  • Don’t overestimate: Setting watermark to "6 hours" when your max delay is 20 minutes unnecessarily increases state size and recovery time after failures.
  • Pair watermark with allowedLateness: Use watermark to bound state growth, and allowedLateness to handle known outliers.
  • Monitor dropped events: Add a foreachBatch sink to log events that fall below the watermark:
stream.writeStream
  .foreachBatch { (batchDF, batchId) =>
    val lateEvents = batchDF.filter(col("eventTime").lt(currentWatermark()))
    if (lateEvents.count() > 0) {
      logWarning(s"Dropped ${lateEvents.count()} late events in batch $batchId")
      lateEvents.write.mode("append").json("/mnt/late-events/")
    }
  }
  .start()
  • Test with deliberate lateness: In your test environment, use a script to inject events with past timestamps and verify they’re processed correctly.

Pitfalls I’ve Encountered

One mistake I made early on was assuming allowedLateness affects the watermark itself. It doesn’t — it only extends the window for state updates. The watermark still advances based on max_event_time_seen - delay_threshold. If you set allowedLateness higher than your watermark delay, you’re not increasing how long you wait — you’re just allowing late updates to existing state.

Another gotcha: if you’re using outputMode("complete"), allowedLateness has no effect because the entire state is recomputed and rewritten each time. Use update or append mode for lateness to work.

Finally, be careful with state cleanup. Spark automatically clears state for watermarked keys, but if you’re using custom state maps or flatMapGroupsWithState, you need to handle eviction yourself based on watermark.

Conclusion

Watermark and allowedLateness aren’t just configuration knobs — they’re essential tools for building reliable event-time pipelines. By setting them based on real observed latency and testing edge cases, I’ve gone from losing double-digit percentages of data to achieving sub-0.1% loss in my Structured Streaming jobs.

If you’re running Spark Structured Streaming in production and haven’t tuned these parameters yet, start by measuring your actual latency. A 15-minute watermark might be all you need — or you might discover, like I did, that your mobile users need a 2-hour window to stay connected. The key is to align your settings with reality, not assumptions.

As I mentioned before in my post about [monitoring streaming lag](https://furkanikkan.com/urun/perfetto-ile-mikro-saniye-cozunurlukte-cekirdek-fonksiyon-cagrilarinin-profili-olusturulmasi-83), observability is half the battle. The other half is designing your stream to be resilient to the messiness of real-world data.

Cover image: USDAgov · PDM (Openverse / kamu malı) · https://www.flickr.com/photos/41284017@N08/54674581419