Watermarking for late data in Apache Spark - Step-by-Step Execution

Concept Flow - Watermarking for late data

Start Streaming Data

↓

Assign Event Time

↓

Set Watermark Threshold

↓

Process Data with Window

↓

Drop Late Data beyond Watermark

↓

Output Cleaned Stream

↓

End

Data flows in with event times; watermark sets a threshold to drop data arriving too late, ensuring timely processing.

Execution Sample

Apache Spark

from pyspark.sql.functions import window

stream.withWatermark("eventTime", "10 minutes")
  .groupBy(window("eventTime", "5 minutes"))
  .count()

This code sets a watermark of 10 minutes on eventTime and counts events in 5-minute windows, dropping data later than 10 minutes.

Execution Table

Step	Input Event Time	Watermark Threshold	Is Late?	Action	Output
1	2024-06-01 10:00:00	None (initial)	No	Accept event	Count updated for window 10:00-10:05
2	2024-06-01 10:04:00	None (initial)	No	Accept event	Count updated for window 10:00-10:05
3	2024-06-01 09:50:00	None (initial)	No	Accept event	Count updated for window 09:50-09:55
4	2024-06-01 10:15:00	10:05:00 (eventTime - 10 min)	No	Accept event	Count updated for window 10:15-10:20
5	2024-06-01 09:52:00	10:05:00	Yes	Drop event (too late)	No change
6	2024-06-01 10:07:00	10:05:00	No	Accept event	Count updated for window 10:05-10:10
7	2024-06-01 09:54:00	10:05:00	Yes	Drop event (too late)	No change
8	2024-06-01 10:12:00	10:05:00	No	Accept event	Count updated for window 10:10-10:15
9	2024-06-01 10:20:00	10:05:00	No	Accept event	Count updated for window 10:20-10:25
10	2024-06-01 09:59:00	10:10:00	Yes	Drop event (too late)	No change
Exit	N/A	N/A	N/A	Stream ends or watermark advances beyond all event times	Final counts for windows

💡 Watermark advances with eventTime minus 10 minutes; events older than watermark are dropped as late.

Variable Tracker

Variable	Start	After Step 4	After Step 6	After Step 9	Final
watermark	None	2024-06-01 10:05:00	2024-06-01 10:05:00	2024-06-01 10:10:00	2024-06-01 10:10:00
accepted_events_count	{}	{"10:00-10:05":2, "09:50-09:55":1, "10:15-10:20":1}	{"10:00-10:05":2, "09:50-09:55":1, "10:05-10:10":1, "10:15-10:20":1}	{"10:00-10:05":2, "09:50-09:55":1, "10:05-10:10":1, "10:10-10:15":1, "10:15-10:20":1, "10:20-10:25":1}	{"10:00-10:05":2, "09:50-09:55":1, "10:05-10:10":1, "10:10-10:15":1, "10:15-10:20":1, "10:20-10:25":1}
dropped_events_count	0	0	1	2	3

Key Moments - 2 Insights

Why are some events with earlier timestamps dropped even though they arrive later?

Because the watermark moves forward with the event time minus the allowed delay (10 minutes). Events older than the watermark are considered too late and dropped, as shown in steps 5, 7, and 10 in the execution_table.

Does the watermark move backward if an older event arrives late?

No, the watermark only moves forward or stays the same. Late events do not move the watermark backward, so they get dropped if older than the watermark, as seen in steps 5 and 7.

Visual Quiz - 3 Questions

Look at the execution_table at step 5. Why is the event with time 09:52:00 dropped?

A Because it arrived before the watermark was set

B Because its event time is older than the watermark 10:05:00

C Because it belongs to a future window

D Because the count for that window is already full

💡 Hint

Check the 'Is Late?' and 'Watermark Threshold' columns at step 5 in the execution_table.

At which step does the watermark first get set to 10:05:00?

A Step 1

B Step 3

C Step 4

D Step 6

💡 Hint

Look at the 'Watermark Threshold' column in the execution_table rows.

If the watermark delay was changed from 10 minutes to 5 minutes, what would happen to the event at 09:54:00 arriving at step 7?

A It would be dropped because the watermark is earlier

B It would be accepted because the watermark is later

C It would update the watermark backward

D It would be counted twice

💡 Hint

Consider how watermark delay affects the threshold for dropping late data, referencing step 7 in the execution_table.

Concept Snapshot

Watermarking for late data in streaming:
- Assign event time to data
- Set watermark delay (e.g., 10 min)
- Watermark = max event time seen - delay
- Drop events older than watermark
- Ensures timely, bounded-lateness processing

Full Transcript

Watermarking helps handle late data in streaming by setting a threshold time. Events with timestamps older than this threshold are dropped to keep processing timely. The watermark moves forward as new data arrives, never backward. This example shows events accepted or dropped based on their event time compared to the watermark, with counts updated for time windows. Late events beyond the watermark are ignored to avoid waiting indefinitely.