Incremental processing patterns in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Output of Spark Structured Streaming with Watermark

Consider the following Spark Structured Streaming code snippet that reads a stream of events with timestamps and applies a watermark to handle late data. What will be the output count of events after the watermark is applied and aggregation is done?

Apache Spark

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

spark = SparkSession.builder.appName('WatermarkExample').getOrCreate()

# Sample streaming DataFrame simulating event data
input_data = [
    ("2024-06-01 10:00:00", "event1"),
    ("2024-06-01 10:01:00", "event2"),
    ("2024-06-01 09:55:00", "event3"),  # Late event
    ("2024-06-01 10:02:00", "event4")
]

schema = "timestamp STRING, event STRING"

# Create static DataFrame to simulate streaming input
static_df = spark.createDataFrame(input_data, schema=schema)

# Convert timestamp string to timestamp type
streaming_df = static_df.withColumn('timestamp', col('timestamp').cast('timestamp'))

# Apply watermark and group by window
agg_df = streaming_df.withWatermark('timestamp', '5 minutes') \
    .groupBy(window(col('timestamp'), '10 minutes')) \
    .count()

result = agg_df.collect()
output = [(row['window'].start.strftime('%Y-%m-%d %H:%M:%S'), row['count']) for row in result]
print(output)

A [('2024-06-01 10:00:00', 4)]

B [('2024-06-01 09:50:00', 4)]

C [('2024-06-01 10:00:00', 3)]

D [('2024-06-01 09:50:00', 3)]

💡 Hint

Remember that watermark drops events older than the watermark threshold relative to the max event time seen.

The watermark of 5 minutes means events older than 5 minutes from the max event timestamp are dropped. The event at 09:55:00 is within 5 minutes of the max timestamp 10:02:00, so it is included. The window groups events from 10:00:00 to 10:10:00, so the count is 3 events in that window.

❓ data_output

intermediate

Resulting DataFrame after Incremental Batch Processing

You have a batch DataFrame with user transactions and a checkpoint storing the last processed transaction ID. You want to process only new transactions with IDs greater than the checkpoint. Given the following code, what will be the resulting DataFrame after filtering?

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName('IncrementalBatch').getOrCreate()

# Sample transactions
transactions = [
    (1, 'user1', 100),
    (2, 'user2', 150),
    (3, 'user3', 200),
    (4, 'user1', 250)
]

schema = 'transaction_id INT, user STRING, amount INT'

df = spark.createDataFrame(transactions, schema=schema)

last_checkpoint = 2

# Filter new transactions
new_transactions = df.filter(col('transaction_id') > last_checkpoint)

result = new_transactions.collect()
output = [(row['transaction_id'], row['user'], row['amount']) for row in result]
print(output)

A [(3, 'user3', 200), (4, 'user1', 250)]

B [(1, 'user1', 100), (2, 'user2', 150)]

C [(4, 'user1', 250)]

D [(2, 'user2', 150), (3, 'user3', 200)]

💡 Hint

Filter selects transactions with ID strictly greater than the checkpoint.

Filtering with transaction_id > 2 selects transactions with IDs 3 and 4 only.

🔧 Debug

advanced

Identify the Error in Incremental Stream Processing Code

The following Spark Structured Streaming code attempts to read a stream, apply a watermark, and write the output to console. However, it raises an error. What is the cause?

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName('StreamError').getOrCreate()

streaming_df = spark.readStream.format('csv').option('header', 'true').load('/path/to/stream')

# Apply watermark on a non-timestamp column
agg_df = streaming_df.withWatermark('event_time', '10 minutes').groupBy('user').count()

query = agg_df.writeStream.outputMode('append').format('console').start()
query.awaitTermination()

A FileNotFoundError: /path/to/stream does not exist

B No error, code runs successfully

C StreamingQueryException: Output mode 'append' not supported with aggregation

D AnalysisException: Watermark can only be applied on timestamp columns

💡 Hint

Check the data type of the column used in withWatermark.

Watermark requires the column to be of timestamp type. If 'event_time' is not timestamp, Spark raises AnalysisException.

❓ visualization

advanced

Visualizing Incremental Data Processing with Micro-batches

You run a Spark Structured Streaming job with micro-batch interval of 5 seconds. The job processes data arriving every 3 seconds. Which visualization best represents the processing timeline of micro-batches and data arrival?

A A timeline showing micro-batches starting every 3 seconds, matching data arrival exactly, with no overlap.

B A timeline showing micro-batches starting every 5 seconds, each processing data that arrived in the previous 5 seconds, with some overlap in data arrival times.

C A timeline showing micro-batches starting every 10 seconds, processing data in 5-second windows, with gaps in processing.

D A timeline showing continuous processing with no discrete micro-batches, processing data instantly as it arrives.

💡 Hint

Micro-batch interval controls how often batches start, independent of data arrival frequency.

Micro-batches start at fixed intervals (5 seconds). Data arriving every 3 seconds accumulates and is processed in these batches, causing some overlap in data arrival times within batches.

🧠 Conceptual

expert

Choosing the Best Incremental Processing Pattern for Late Data

You have a streaming data source with frequent late-arriving events. You want to ensure accurate aggregation results while minimizing state size and latency. Which incremental processing pattern is best suited for this scenario?

A Use event-time based watermarking with allowed lateness and state cleanup to handle late data efficiently.

B Use processing-time triggers without watermarking to process data as soon as it arrives, ignoring late events.

C Use batch processing only, ignoring streaming capabilities to avoid late data complexity.

D Use windowed aggregation without watermarking and keep all state indefinitely to capture all late events.

💡 Hint

Consider how watermarking helps manage late data and state size.

Watermarking with allowed lateness lets Spark drop very late events and clean up state, balancing accuracy and resource use. Processing-time triggers or no watermarking either ignore late data or cause unbounded state growth.

Practice

1.

What is the main idea behind incremental processing in Apache Spark?

easy

A. Delete old data before processing new data

B. Process all data repeatedly for accuracy

C. Process only new or changed data instead of all data every time

D. Use only batch processing without updates

Solution

1. Step 1: Understand incremental processing concept

   Incremental processing means working only on data that is new or has changed, not the entire dataset.

2. Step 2: Compare options to concept

   Process only new or changed data instead of all data every time matches this idea, while others describe full or batch processing.

3. Final Answer:

   Process only new or changed data instead of all data every time -> Option C

4. Quick Check:

   Incremental processing = new or changed data only [OK]

Hint: Incremental means update only new or changed data [OK]

Common Mistakes:

• Thinking incremental means reprocessing all data
• Confusing incremental with deleting old data
• Assuming incremental ignores changes

2.

Which Spark code snippet correctly filters new data for incremental processing?

df_new = df_all.filter(____)

easy

A. df_all.timestamp > last_processed_time

B. df_all.timestamp < last_processed_time

C. df_all.id == last_processed_id

D. df_all.date == '2020-01-01'

Solution

1. Step 1: Identify filtering condition for new data

   To get new data, filter rows with timestamp greater than last processed time.

2. Step 2: Match condition to options

   df_all.timestamp > last_processed_time uses > last_processed_time, which correctly selects new data.

3. Final Answer:

   df_all.timestamp > last_processed_time -> Option A

4. Quick Check:

   Filter new data by timestamp > last processed [OK]

Hint: Filter with timestamp greater than last processed time [OK]

Common Mistakes:

• Using less than instead of greater than
• Filtering by exact id instead of timestamp
• Filtering by unrelated fixed date

3.

Given this Spark code for incremental processing, what will df_incremental.count() output?

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
data = [(1, '2024-01-01'), (2, '2024-01-02'), (3, '2024-01-03')]
df_all = spark.createDataFrame(data, ['id', 'date'])
last_processed_date = '2024-01-01'
df_incremental = df_all.filter(df_all.date > last_processed_date)
print(df_incremental.count())

medium

A. 2

B. 3

C. 1

D. 0

Solution

1. Step 1: Understand filtering condition

   Filter keeps rows where date > '2024-01-01'. Data has dates '2024-01-01', '2024-01-02', '2024-01-03'.

2. Step 2: Count rows matching condition

   Rows with dates '2024-01-02' and '2024-01-03' match, so count is 2.

3. Final Answer:

   2 -> Option A

4. Quick Check:

   Dates after '2024-01-01' = 2 rows [OK]

Hint: Count rows with date greater than last processed date [OK]

Common Mistakes:

• Including the date equal to last_processed_date
• Counting all rows instead of filtered
• Confusing greater than with less than

4.

Find the error in this Spark code for incremental processing:

df_incremental = df_all.filter(df_all.timestamp < last_processed_time)

Assuming last_processed_time is the timestamp of the last processed batch.

medium

A. The variable last_processed_time is undefined

B. The filter should use <= to include last processed data

C. The filter should use == to match last processed time

D. The filter should use > instead of < to get new data

Solution

1. Step 1: Understand filtering logic for incremental data

   Incremental processing requires filtering data newer than last processed time, so timestamp > last_processed_time.

2. Step 2: Identify incorrect operator

   The code uses < which selects older data, not new data.

3. Final Answer:

   The filter should use > instead of < to get new data -> Option D

4. Quick Check:

   New data filter uses > last processed time [OK]

Hint: New data means timestamp greater than last processed [OK]

Common Mistakes:

• Using less than instead of greater than
• Confusing equality with incremental filter
• Assuming variable is undefined without checking

5.

You have a Delta Lake table storing daily sales data. You want to update it incrementally with new data from a streaming source. Which approach is best?

hard

A. Delete the Delta table and recreate it daily

B. Use merge operation to upsert new and updated records into the Delta table

C. Append new data without checking for duplicates or updates

D. Overwrite the entire Delta table with new streaming data each time

Solution

1. Step 1: Understand Delta Lake incremental update capabilities

   Delta Lake supports merge to upsert (insert or update) records efficiently.

2. Step 2: Evaluate options for incremental update

   Use merge operation to upsert new and updated records into the Delta table uses merge, which handles new and changed data correctly. Others cause data loss or duplicates.

3. Final Answer:

   Use merge operation to upsert new and updated records into the Delta table -> Option B

4. Quick Check:

   Delta Lake incremental update = merge upsert [OK]

Hint: Use merge for incremental upsert in Delta Lake [OK]

Common Mistakes:

• Overwriting entire table loses old data
• Appending without deduplication causes duplicates
• Deleting table loses all history