Watermarking for late data in Apache Spark - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to set a watermark on the streaming DataFrame to handle late data.

Apache Spark

streamingDF = streamingDF.withWatermark('[1]', '10 minutes')

Drag options to blanks, or click blank then click option'

Atimestamp

BeventTime

Cdate

Dtime

Attempts:

3 left

💡 Hint

Common Mistakes

Using a column name that does not exist in the DataFrame.

Using a non-timestamp column for watermarking.

The 'timestamp' column is commonly used as the event time column for watermarking in Spark Structured Streaming.

2fill in blank

medium

Complete the code to drop late data that is older than the watermark delay.

Apache Spark

cleanedDF = streamingDF.dropDuplicates(['userId', '[1]'])

Drag options to blanks, or click blank then click option'

Adate

Btimestamp

CeventTime

Dtime

Attempts:

3 left

💡 Hint

Common Mistakes

Using a non-unique column for dropping duplicates.

Using a column different from the watermark timestamp.

Dropping duplicates based on 'userId' and 'timestamp' helps remove late data beyond the watermark.

3fill in blank

hard

Fix the error in the watermarking code by completing the missing argument.

Apache Spark

streamingDF = streamingDF.withWatermark('timestamp', '[1]')

Drag options to blanks, or click blank then click option'

A10 minutes

B10min

Cminutes 10

D10

Attempts:

3 left

💡 Hint

Common Mistakes

Using only a number without units.

Using incorrect unit format.

The watermark delay must be a string with a number and a time unit, e.g., '10 minutes'.

4fill in blank

hard

Fill both blanks to create a streaming aggregation with watermarking and windowing.

Apache Spark

result = streamingDF.withWatermark('[1]', '5 minutes').groupBy(window('[2]', '10 minutes')).count()

Drag options to blanks, or click blank then click option'

Atimestamp

BeventTime

Attempts:

3 left

💡 Hint

Common Mistakes

Using different columns for watermark and window.

Using a non-timestamp column.

Both watermark and window functions should use the 'timestamp' column for consistency.

5fill in blank

hard

Fill all three blanks to filter late data using watermark and event time comparison.

Apache Spark

filteredDF = streamingDF.withWatermark('[1]', '15 minutes').filter(streamingDF.[2] > streamingDF.[3])

Drag options to blanks, or click blank then click option'

Atimestamp

BeventTime

Cwatermark

Attempts:

3 left

💡 Hint

Common Mistakes

Using inconsistent column names.

Comparing wrong columns in filter.

Watermarking uses 'timestamp' column; filtering compares 'timestamp' to 'watermark' to exclude late data.

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