Watermarking for late data in Apache Spark - Mini Project: Build & Apply

Watermarking for late data

📖 Scenario: You work for a company that processes streaming data of user clicks on a website. Sometimes, data arrives late due to network delays. To handle this, you want to use watermarking to ignore very late data and keep your analysis accurate.

🎯 Goal: Build a Spark Structured Streaming job that reads click events, applies watermarking on event time to handle late data, and counts clicks per user in a time window.

📋 What You'll Learn

Create a streaming DataFrame with sample click data including event timestamps

Set a watermark on the event time column with a delay threshold

Group data by user and time window to count clicks

Output the aggregated counts

💡 Why This Matters

🌍 Real World

Watermarking helps streaming systems ignore very late data that can cause incorrect results, improving data accuracy in real-time analytics.

💼 Career

Data engineers and data scientists use watermarking in Spark Structured Streaming to manage late-arriving data in event-time processing pipelines.

1

Create streaming DataFrame with click data

Create a Spark streaming DataFrame called clicks with the following schema and data: columns user (string) and event_time (timestamp). Use the following data rows: ("Alice", "2024-06-01 12:00:00"), ("Bob", "2024-06-01 12:01:00"), ("Alice", "2024-06-01 12:02:00"). Use spark.createDataFrame and to_timestamp to convert strings to timestamps.

Apache Spark

# Create the streaming DataFrame called clicks with user and event_time columns
# Your code here

Hint

Use spark.createDataFrame with a list of tuples and column names. Then use withColumn and to_timestamp to convert the string column to timestamp.

2

Set watermark delay threshold

Create a variable called watermark_delay and set it to the string "1 minute". This will be the delay threshold for watermarking.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import to_timestamp

spark = SparkSession.builder.appName("WatermarkExample").getOrCreate()
data = [("Alice", "2024-06-01 12:00:00"), ("Bob", "2024-06-01 12:01:00"), ("Alice", "2024-06-01 12:02:00")]
clicks = spark.createDataFrame(data, ["user", "event_time_str"]).withColumn("event_time", to_timestamp("event_time_str")).drop("event_time_str")

# Create watermark_delay variable with value "1 minute"
# Your code here

Hint

Just create a variable named watermark_delay and assign the string "1 minute".

3

Apply watermark and group by user and window

Use the clicks DataFrame and apply withWatermark on the event_time column with the delay threshold watermark_delay. Then group by user and a 2-minute time window on event_time using window. Aggregate by counting the number of clicks per group and save the result in a DataFrame called click_counts. Import window from pyspark.sql.functions.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import to_timestamp

spark = SparkSession.builder.appName("WatermarkExample").getOrCreate()
data = [("Alice", "2024-06-01 12:00:00"), ("Bob", "2024-06-01 12:01:00"), ("Alice", "2024-06-01 12:02:00")]
clicks = spark.createDataFrame(data, ["user", "event_time_str"]).withColumn("event_time", to_timestamp("event_time_str")).drop("event_time_str")

watermark_delay = "1 minute"

# Import window function
from pyspark.sql.functions import window

# Apply watermark and group by user and 2-minute window, count clicks
# Your code here

Hint

Use withWatermark on event_time with watermark_delay. Then group by user and window(event_time, "2 minutes"). Finally, use count() to aggregate.

4

Show the aggregated click counts

Use click_counts.show() to display the aggregated counts of clicks per user and time window.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import to_timestamp, window

spark = SparkSession.builder.appName("WatermarkExample").getOrCreate()
data = [("Alice", "2024-06-01 12:00:00"), ("Bob", "2024-06-01 12:01:00"), ("Alice", "2024-06-01 12:02:00")]
clicks = spark.createDataFrame(data, ["user", "event_time_str"]).withColumn("event_time", to_timestamp("event_time_str")).drop("event_time_str")

watermark_delay = "1 minute"

click_counts = clicks.withWatermark("event_time", watermark_delay).groupBy("user", window("event_time", "2 minutes")).count()

# Show the click_counts DataFrame
# Your code here

Hint

Use click_counts.show() to print the result table.

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