Error handling and dead letter queues in Apache Spark - Step-by-Step Execution

Concept Flow - Error handling and dead letter queues

Start Data Processing

↓

Process Each Record

↓

Error Occurs?

No→Write to Output

Yes↓

Send to Dead Letter Queue

↓

Continue Processing Next Record

↓

End

Data is processed record by record; if an error happens, the record is sent to a dead letter queue, else it goes to output.

Execution Sample

Apache Spark

from pyspark.sql.functions import col

def process_record(row):
    try:
        # simulate processing
        val = int(row['value'])
        return (row['id'], val * 2)
    except Exception:
        return None

# Filter out None and send errors to dead letter queue

This code tries to process each record; if it fails, it returns None to mark error for dead letter queue.

Execution Table

Step	Record	Processing Result	Error?	Action	Output	Dead Letter Queue
1	{"id": 1, "value": "10"}	(1, 20)	No	Write to Output	(1, 20)	None
2	{"id": 2, "value": "abc"}	None	Yes	Send to Dead Letter Queue	None	{"id": 2, "value": "abc"}
3	{"id": 3, "value": "5"}	(3, 10)	No	Write to Output	(3, 10)	None
4	{"id": 4, "value": null}	None	Yes	Send to Dead Letter Queue	None	{"id": 4, "value": null}
5	{"id": 5, "value": "7"}	(5, 14)	No	Write to Output	(5, 14)	None
6	End of Data	N/A	N/A	Stop Processing	Final Output Collected	Dead Letter Queue Collected

💡 All records processed; errors sent to dead letter queue; successful processed records written to output.

Variable Tracker

Variable	Start	After 1	After 2	After 3	After 4	After 5	Final
current_record	None	{"id":1,"value":"10"}	{"id":2,"value":"abc"}	{"id":3,"value":"5"}	{"id":4,"value":null}	{"id":5,"value":"7"}	End of Data
processing_result	None	(1, 20)	None	(3, 10)	None	(5, 14)	N/A
dead_letter_queue	[]	[]	[{"id":2,"value":"abc"}]	[{"id":2,"value":"abc"}]	[{"id":2,"value":"abc"},{"id":4,"value":null}]	[{"id":2,"value":"abc"},{"id":4,"value":null}]	[{"id":2,"value":"abc"},{"id":4,"value":null}]
output	[]	[(1,20)]	[(1,20)]	[(1,20),(3,10)]	[(1,20),(3,10)]	[(1,20),(3,10),(5,14)]	[(1,20),(3,10),(5,14)]

Key Moments - 3 Insights

Why do some records go to the dead letter queue instead of output?

Records that cause errors during processing (like invalid data) are caught and sent to the dead letter queue as shown in execution_table rows 2 and 4.

What happens to processing after an error record is found?

Processing continues with the next record without stopping, ensuring all data is handled, as seen in execution_table rows 2 to 5.

How do we know which records failed processing?

Failed records appear in the dead letter queue variable tracked in variable_tracker, accumulating error records step by step.

Visual Quiz - 3 Questions

Look at the execution_table at step 4, what is the processing result and where does the record go?

A Processing result is None, record goes to dead letter queue

B Processing result is (4, null), record goes to output

C Processing result is (4, 0), record goes to output

D Processing result is None, record is discarded

💡 Hint

Check execution_table row 4 columns 'Processing Result' and 'Dead Letter Queue'

At which step does the dead letter queue first receive a record?

A Step 1

B Step 2

C Step 3

D Step 5

💡 Hint

Look at variable_tracker row for 'dead_letter_queue' after each step

If the record at step 3 had an error, how would the output and dead letter queue change?

A Output would include (3, 10), dead letter queue unchanged

B Output adds (3, 10), dead letter queue adds record at step 3

C Output unchanged, dead letter queue adds record at step 3

D Both output and dead letter queue unchanged

💡 Hint

Refer to how errors affect output and dead letter queue in execution_table and variable_tracker

Concept Snapshot

Error handling in Spark processes each record.
If processing fails, record goes to dead letter queue.
Successful records go to output.
Processing continues without stopping on errors.
Dead letter queue helps track problem data for later review.

Full Transcript

This visual execution shows how Apache Spark handles errors during data processing using dead letter queues. Each record is processed one by one. If processing succeeds, the result is written to output. If an error occurs, the record is sent to a dead letter queue instead of stopping the whole job. The execution table tracks each step, showing which records succeed or fail and where they go. The variable tracker shows how the output and dead letter queue grow over time. Key moments clarify why errors are handled this way and how processing continues smoothly. The quiz tests understanding by asking about specific steps and changes in output or dead letter queue. This approach helps keep data pipelines robust and easy to debug.

Practice

1. What is the main purpose of using a dead letter queue in Apache Spark data pipelines?

easy

A. To speed up the processing of good data by caching it

B. To store bad or invalid data separately for later analysis or correction

C. To automatically fix errors in the data during processing

D. To merge multiple data sources into one unified dataset

Solution

1. Step 1: Understand the role of error handling in data pipelines

   Error handling separates good data from bad data to prevent pipeline failures.

2. Step 2: Define the purpose of a dead letter queue

   A dead letter queue stores bad or invalid data separately so it can be reviewed or fixed later without stopping the main pipeline.

3. Final Answer:

   To store bad or invalid data separately for later analysis or correction -> Option B

4. Quick Check:

   Dead letter queue = store bad data separately [OK]

Hint: Dead letter queue holds bad data separately [OK]

Common Mistakes:

• Thinking dead letter queue fixes data automatically
• Confusing dead letter queue with caching good data
• Assuming it merges data sources

2. Which of the following Spark code snippets correctly filters out invalid records and writes them to a dead letter queue DataFrame named dlqDF?

easy

A. val dlqDF = inputDF.filter(col("age") < 0)

B. val dlqDF = inputDF.select("age")

C. val dlqDF = inputDF.filter(col("age") > 0)

D. val dlqDF = inputDF.drop("age")

Solution

1. Step 1: Identify the condition for invalid data

   Invalid records have age less than 0, so filter with col("age") < 0.

2. Step 2: Check which code filters invalid data correctly

   val dlqDF = inputDF.filter(col("age") < 0) filters rows where age is less than 0, capturing invalid data for the dead letter queue.

3. Final Answer:

   val dlqDF = inputDF.filter(col("age") < 0) -> Option A

4. Quick Check:

   Filter invalid age < 0 = val dlqDF = inputDF.filter(col("age") < 0) [OK]

Hint: Filter invalid rows with condition in filter() [OK]

Common Mistakes:

• Filtering valid data instead of invalid
• Using select instead of filter
• Dropping columns instead of filtering rows

3. Given the following Spark code, what will be the count of records in dlqDF?

val inputDF = spark.createDataFrame(Seq((1, 25), (2, -5), (3, 30), (4, -1))).toDF("id", "age")
val dlqDF = inputDF.filter(col("age") < 0)
dlqDF.count()

medium

A. 2

B. 4

C. 0

D. 3

Solution

1. Step 1: Identify records with age less than 0

   Records with age -5 and -1 meet the condition, so 2 records are invalid.

2. Step 2: Count filtered records in dlqDF

   dlqDF contains these 2 invalid records, so count is 2.

3. Final Answer:

   2 -> Option A

4. Quick Check:

   Count invalid age < 0 = 2 [OK]

Hint: Count rows where age < 0 to find dlq size [OK]

Common Mistakes:

• Counting all rows instead of filtered
• Miscounting invalid records
• Confusing filter condition direction

4. The following Spark code is intended to separate invalid records (age < 0) into a dead letter queue DataFrame dlqDF. However, it does not work as expected. What is the error?

val dlqDF = inputDF.filter("age > 0")

medium

A. The DataFrame name dlqDF is reserved and cannot be used

B. The filter syntax is incorrect; it should use col() function

C. The filter method cannot be used with string conditions

D. The filter condition is reversed; it selects valid data instead of invalid

Solution

1. Step 1: Analyze the filter condition

   The condition "age > 0" selects records with positive age, which are valid, not invalid.

2. Step 2: Identify the correct condition for invalid data

   Invalid data has age less than 0, so the filter should be "age < 0" to capture bad records.

3. Final Answer:

   The filter condition is reversed; it selects valid data instead of invalid -> Option D

4. Quick Check:

   Filter condition reversed = The filter condition is reversed; it selects valid data instead of invalid [OK]

Hint: Check filter logic matches invalid data condition [OK]

Common Mistakes:

• Using wrong comparison operator in filter
• Thinking string filter syntax is invalid
• Assuming DataFrame names are restricted

5. You have a Spark DataFrame inputDF with a column status that can be 'valid' or 'invalid'. You want to create two DataFrames: cleanDF with only valid records and dlqDF with invalid records. Which code correctly implements this error handling with dead letter queue?

hard

A. val cleanDF = inputDF.filter(col("status") = "valid")
val dlqDF = inputDF.filter(col("status") = "invalid")

B. val cleanDF = inputDF.select("status" == "valid")
val dlqDF = inputDF.select("status" == "invalid")

C. val cleanDF = inputDF.filter(col("status") === "valid")
val dlqDF = inputDF.filter(col("status") === "invalid")

D. val cleanDF = inputDF.drop("invalid")
val dlqDF = inputDF.drop("valid")

Solution

1. Step 1: Use correct equality operator in filter

   In Spark Scala, use triple equals (===) to compare column values to strings.

2. Step 2: Filter valid and invalid records separately

   Filter status === "valid" for cleanDF and status === "invalid" for dlqDF to separate data correctly.

3. Final Answer:

   val cleanDF = inputDF.filter(col("status") === "valid") val dlqDF = inputDF.filter(col("status") === "invalid") -> Option C

4. Quick Check:

   Use === for equality and filter separately [OK]

Hint: Use === for string equality in filter() [OK]

Common Mistakes:

• Using single = instead of === in filter
• Using select instead of filter for conditions
• Dropping columns instead of filtering rows