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.

Practice

1. What is the main purpose of watermarking in Apache Spark streaming?

easy

A. To handle late data by setting a threshold time for event processing

B. To speed up the streaming job by skipping data

C. To permanently store all incoming data

D. To convert batch data into streaming data

Solution

1. Step 1: Understand watermarking concept

   Watermarking sets a time limit to wait for late data in streaming.

2. Step 2: Identify purpose in streaming

   It helps to ignore data that arrives too late to keep results timely.

3. Final Answer:

   To handle late data by setting a threshold time for event processing -> Option A

4. Quick Check:

   Watermarking = handle late data [OK]

Hint: Watermarking sets max wait time for late events [OK]

Common Mistakes:

• Thinking watermark speeds up jobs by skipping all data
• Confusing watermark with data storage
• Assuming watermark converts batch to streaming

2. Which of the following is the correct syntax to add a watermark on a streaming DataFrame df with event time column eventTime and delay threshold of 10 minutes?

easy

A. df.withWatermark(eventTime, 10)

B. df.withWatermark("eventTime", "10 minutes")

C. df.watermark("eventTime", 10)

D. df.withWatermark("eventTime", 10)

Solution

1. Step 1: Recall correct method signature

   The method is withWatermark with two string arguments: column name and delay duration.

2. Step 2: Match correct syntax

   df.withWatermark("eventTime", "10 minutes") uses correct method and string format for delay "10 minutes".

3. Final Answer:

   df.withWatermark("eventTime", "10 minutes") -> Option B

4. Quick Check:

   Correct syntax uses strings for both args [OK]

Hint: Use strings for column and delay in withWatermark() [OK]

Common Mistakes:

• Passing column name without quotes
• Using integer instead of string for delay
• Using wrong method name like watermark()

3. Given the following Spark Structured Streaming code snippet, what will be the output schema's watermark column after applying watermark?

val streamingDF = spark.readStream.format("csv")
  .option("header", "true")
  .load("/data/events")

val withWatermarkDF = streamingDF.withWatermark("eventTime", "5 minutes")
withWatermarkDF.printSchema()

medium

A. eventTime: timestamp (nullable = true) with watermark

B. eventTime: string (nullable = true) with watermark

C. eventTime: timestamp (nullable = false) without watermark

D. No watermark added to schema

Solution

1. Step 1: Understand watermark effect on schema

   Watermark adds metadata to the event time column, which must be timestamp type.

2. Step 2: Check expected schema output

   Schema shows eventTime as timestamp with watermark metadata.

3. Final Answer:

   eventTime: timestamp (nullable = true) with watermark -> Option A

4. Quick Check:

   Watermark adds metadata to timestamp column [OK]

Hint: Watermark adds metadata to timestamp event column [OK]

Common Mistakes:

• Assuming eventTime remains string type
• Thinking watermark changes nullability
• Believing watermark does not affect schema

4. Identify the error in the following Spark Structured Streaming code that uses watermarking:

val df = spark.readStream.format("json")
  .load("/stream/data")

val result = df.withWatermark("timestamp", "1 minute")
  .groupBy(window(col("timestamp"), "30 seconds"))
  .count()

medium

A. withWatermark must be called after groupBy

B. Column name "timestamp" should be "eventTime"

C. Window duration cannot be "30 seconds"

D. Watermark delay must be shorter than window duration

Solution

1. Step 1: Check watermark and window durations

   Watermark delay must be shorter than window duration to ensure the query emits results.

2. Step 2: Compare durations in code

   Watermark is 1 minute but window is 30 seconds, watermark delay longer than window duration and causes error.

3. Final Answer:

   Watermark delay must be shorter than window duration -> Option D

4. Quick Check:

   Watermark < window duration [OK]

Hint: Watermark delay must be < window duration [OK]

Common Mistakes:

• Assuming any column name works
• Thinking window duration cannot be "30 seconds"
• Calling withWatermark after groupBy

5. You have a streaming dataset with event times and want to aggregate counts per 10-minute window. Late data can arrive up to 15 minutes late. Which watermark delay should you set to ensure correct results without excessive delay?

hard

A. "25 minutes"

B. "10 minutes"

C. "15 minutes"

D. "5 minutes"

Solution

1. Step 1: Understand watermark delay role

   Watermark delay should cover the maximum expected lateness of data to include late events.

2. Step 2: Match delay to late data arrival

   Since late data can arrive up to 15 minutes late, watermark delay must be at least 15 minutes.

3. Step 3: Avoid setting delay too short or too long

   Setting delay less than 15 minutes risks dropping late data; too long delays output.

4. Final Answer:

   "15 minutes" -> Option C

5. Quick Check:

   Watermark delay = max late arrival [OK]

Hint: Set watermark delay to max expected late arrival [OK]

Common Mistakes:

• Setting delay equal to window size instead of late arrival
• Using shorter delay than late data time
• Using unnecessarily long delay causing latency