Error handling and dead letter queues in Apache Spark - Mini Project: Build & Apply

Error handling and dead letter queues

📖 Scenario: You work at a company that processes customer orders using Apache Spark. Sometimes, the order data has errors like missing fields or wrong data types. To keep the main data clean, you want to separate the bad records into a special place called a dead letter queue for later review.

🎯 Goal: Build a Spark program that reads a list of orders, checks for errors, separates good orders from bad orders, and shows both results. This helps keep your data clean and makes it easy to fix errors later.

📋 What You'll Learn

Create an initial list of orders with some correct and some incorrect entries

Set a configuration variable for the minimum valid order amount

Use Spark to separate good orders from bad orders based on validation rules

Print the good orders and the dead letter queue (bad orders) separately

💡 Why This Matters

🌍 Real World

In real companies, data often has errors. Separating bad data into a dead letter queue helps keep the main data clean and makes fixing errors easier.

💼 Career

Data engineers and data scientists use error handling and dead letter queues to build reliable data pipelines and maintain data quality.

1

Create the initial orders data

Create a list called orders with these exact dictionaries representing orders: {'order_id': 1, 'customer': 'Alice', 'amount': 250}, {'order_id': 2, 'customer': 'Bob', 'amount': -50}, {'order_id': 3, 'customer': 'Charlie', 'amount': 100}, and {'order_id': 4, 'customer': null, 'amount': 300}.

Apache Spark

# Create the list called orders with the given dictionaries
# Your code here

Hint

Use a Python list with dictionaries exactly as shown.

2

Set the minimum valid order amount

Create a variable called min_valid_amount and set it to 0 to represent the minimum allowed order amount.

Apache Spark

orders = [
    {'order_id': 1, 'customer': 'Alice', 'amount': 250},
    {'order_id': 2, 'customer': 'Bob', 'amount': -50},
    {'order_id': 3, 'customer': 'Charlie', 'amount': 100},
    {'order_id': 4, 'customer': None, 'amount': 300}
]
# Set min_valid_amount to 0
# Your code here

Hint

Just assign 0 to the variable min_valid_amount.

3

Separate good and bad orders using Spark

Use Spark to create a DataFrame from orders. Then create two DataFrames: good_orders with orders where customer is not null and amount is greater than or equal to min_valid_amount, and dead_letter_queue with the rest of the orders.

Apache Spark

orders = [
    {'order_id': 1, 'customer': 'Alice', 'amount': 250},
    {'order_id': 2, 'customer': 'Bob', 'amount': -50},
    {'order_id': 3, 'customer': 'Charlie', 'amount': 100},
    {'order_id': 4, 'customer': None, 'amount': 300}
]

min_valid_amount = 0

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('ErrorHandling').getOrCreate()

# Create DataFrame from orders
# Create good_orders DataFrame with valid orders
# Create dead_letter_queue DataFrame with invalid orders
# Your code here

Hint

Use spark.createDataFrame(orders) to make the DataFrame. Use filter with conditions on customer and amount.

4

Print the good orders and dead letter queue

Use show() to print the good_orders DataFrame and then the dead_letter_queue DataFrame.

Apache Spark

orders = [
    {'order_id': 1, 'customer': 'Alice', 'amount': 250},
    {'order_id': 2, 'customer': 'Bob', 'amount': -50},
    {'order_id': 3, 'customer': 'Charlie', 'amount': 100},
    {'order_id': 4, 'customer': None, 'amount': 300}
]

min_valid_amount = 0

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('ErrorHandling').getOrCreate()

orders_df = spark.createDataFrame(orders)
good_orders = orders_df.filter((orders_df.customer.isNotNull()) & (orders_df.amount >= min_valid_amount))
dead_letter_queue = orders_df.filter((orders_df.customer.isNull()) | (orders_df.amount < min_valid_amount))

# Print good_orders
# Print dead_letter_queue
# Your code here

Hint

Use good_orders.show() and dead_letter_queue.show() to display the results.

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