Watermarks and late data for Spark & Data Eng developers

Why watermarks matter in streaming

In streaming systems, data doesn't always arrive in perfect order. A user action logged at 3:00 PM might reach your Spark pipeline at 3:05 PM due to network delays, device connectivity issues, or queueing in upstream systems. Without a mechanism to handle this lateness, you risk either dropping legitimate events or holding state open indefinitely, burning memory and compute. Watermarks and late data for Spark & Data Eng developers is the pattern you need to master to balance correctness against resource efficiency.

This is day 36 of our Spark streaming deep-dive, and we're tackling a problem that feels simple until you encounter it in production: how do you know when to stop waiting for stragglers and finalize a window of aggregation?

withWatermark sets lateness tolerance

The withWatermark API tells Spark how long you're willing to wait for late data before you consider a window "closed." You specify both an event-time column and a delay threshold.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, window, count

spark = SparkSession.builder.appName("WatermarkDemo").getOrCreate()

# Simulated stream of events with event_time and user_id
events_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "events") \
    .load()

parsed = events_df.select(
    col("value").cast("string"),
    col("timestamp").cast("timestamp").alias("event_time")
)

# Set watermark: tolerate up to 10 minutes of lateness
windowed = parsed \
    .withWatermark("event_time", "10 minutes") \
    .groupBy(window(col("event_time"), "5 minutes")) \
    .agg(count("*").alias("event_count"))

query = windowed.writeStream \
    .format("console") \
    .option("checkpointLocation", "/tmp/checkpoint") \
    .start()

query.awaitTermination()

In this example, we define a 5-minute tumbling window on event times, but we tell Spark to keep that window open for an additional 10 minutes. If an event arrives with an event_time that falls into a window that closed more than 10 minutes ago (in terms of the watermark's progress), it will be dropped.

The watermark moves forward automatically as Spark processes newer events. When the watermark for a given window exceeds the window's end time plus the delay threshold, that window is considered complete and won't accept further updates.

State is cleaned after the watermark

One of the most valuable properties of watermarks is automatic state cleanup. In stateful operations—joins, aggregations, deduplication—Spark keeps intermediate results in memory. Without watermarks, that state grows unbounded.

Consider a typical session window where you aggregate user actions within 30 minutes of inactivity:

from pyspark.sql.functions import session_window

sessions = parsed \
    .withWatermark("event_time", "15 minutes") \
    .groupBy(
        col("user_id"),
        session_window(col("event_time"), "30 minutes")
    ) \
    .agg(count("*").alias("action_count"))

query = sessions.writeStream \
    .format("console") \
    .option("checkpointLocation", "/tmp/session_checkpoint") \
    .start()

Behind the scenes, Spark maintains a state store that tracks which sessions are active. For each user, it holds data about their recent actions. When the watermark advances past a session's logical end time plus the tolerance (15 minutes in this case), Spark automatically removes that session's state from memory.

Without this cleanup, a long-running stream would consume more and more memory until the system crashes. The watermark is your contract with Spark: "I promise not to send you events older than this threshold, so you can safely discard state older than that."

This is why choosing your watermark carefully is critical. If you set it too low, you'll drop legitimate late data and produce incorrect results. If you set it too high, you'll consume excessive memory and delay window finalization.

Choose your watermark based on real delays

The watermark value isn't arbitrary. It should reflect the maximum acceptable lateness in your data pipeline, informed by production telemetry and SLA requirements.

Start by instrumenting your ingestion layer to measure how late events actually arrive. Log the difference between event time and processing time for each record:

from pyspark.sql.functions import current_timestamp

# Add processing time to track lateness
with_lateness = parsed.select(
    "*",
    (current_timestamp() - col("event_time")).alias("lateness_seconds")
)

# Write to a metrics topic or sink to understand distribution
lateness_stats = with_lateness \
    .writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "lateness_metrics") \
    .option("checkpointLocation", "/tmp/lateness_checkpoint") \
    .start()

Analyze this data over a week or month to understand your lateness distribution. A typical e-commerce or SaaS platform might see 95% of events arrive within 2 minutes, 99% within 10 minutes, and the occasional outlier at 30+ minutes.

Your watermark should cover the lateness percentile that aligns with your business requirements. If you can tolerate 1% data loss from the very latest devices or networks, set your watermark to the 99th percentile. If accuracy is critical and you can afford the memory overhead, go higher—perhaps 30 or 45 minutes.

For fraud detection systems where every transaction matters, you might use a 60-minute watermark. For analytical dashboards updated hourly, 10 minutes might suffice. The choice is data-driven and business-driven.

Late events beyond the watermark are dropped

Once an event's event-time falls outside the watermark window, Spark silently drops it. This happens at the aggregation or join level, not before ingestion.

Consider a join between two streams:

stream_a = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "stream_a") \
    .load()

stream_b = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "stream_b") \
    .load()

a_parsed = stream_a.select(
    col("value").cast("string").alias("key_a"),
    col("timestamp").cast("timestamp").alias("time_a")
)

b_parsed = stream_b.select(
    col("value").cast("string").alias("key_b"),
    col("timestamp").cast("timestamp").alias("time_b")
)

# Watermark on both streams
a_watermarked = a_parsed.withWatermark("time_a", "5 minutes")
b_watermarked = b_parsed.withWatermark("time_b", "5 minutes")

# Stream-stream join
joined = a_watermarked.join(
    b_watermarked,
    expr("key_a = key_b AND ABS(time_a - time_b) < interval '1 minute'"),
    "inner"
)

query = joined.writeStream \
    .format("console") \
    .option("checkpointLocation", "/tmp/join_checkpoint") \
    .start()

If an event from stream A with time_a at 10:00 arrives at 10:06, it's outside the 5-minute watermark. It cannot be joined with events from stream B that arrived in the normal flow, and it will be dropped. Any join state kept for that key is also discarded.

This dropping behavior is deterministic and logged. You should monitor how many late records are being discarded. A sudden spike in late drops might indicate a problem upstream (a service is slower than expected) or that your watermark is too aggressive.

Common mistakes

• Setting the watermark in the wrong place: Apply withWatermark immediately after parsing event time, before your aggregation or join. Applying it after a stateless operation might not trigger cleanup effectively.
• Confusing watermark with max delay: The watermark is a tolerance, not a hard guarantee. Events can arrive late and be processed if they arrive before the watermark moves past them. Don't rely on watermarks for data validation.
• Using processing time instead of event time: Always watermark on the business-relevant event time, not the time the record hit your cluster. Processing time varies with cluster load and network conditions and doesn't reflect real-world delays in your data.
• Ignoring state cleanup costs: A watermark that's too permissive will keep memory usage high. Monitor your state store size and adjust downward if acceptable lateness increases state size unsustainably.
• Forgetting to checkpoint: Watermark state is part of your streaming checkpoint. Without proper checkpointing, restarting your application will lose watermark progress, potentially causing data loss or duplication.

Key takeaways

• Use withWatermark to define how long Spark should hold state and accept late data for a given column and tolerance window.
• Watermarks enable automatic state cleanup, preventing memory exhaustion in long-running streams.
• Choose your watermark threshold based on actual lateness measurements from your production pipeline, balanced against memory and accuracy requirements.
• Events arriving after the watermark for their window are silently dropped; monitor late-drop metrics to detect upstream issues.
• Always apply watermarks to event time, not processing time, and place them early in your pipeline before stateful operations.