Error handling and dead letter queues in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Spark DataFrame error handling code?

Consider the following PySpark code snippet that tries to parse a JSON column and handle errors by sending bad records to a dead letter queue DataFrame. What will be the count of records in the dead letter queue DataFrame?

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StructField, StringType

spark = SparkSession.builder.appName('ErrorHandling').getOrCreate()
data = [
    ('{"name":"Alice","age":"30"}',),
    ('{"name":"Bob","age":"not_a_number"}',),
    ('{"name":"Charlie","age":"25"}',),
    ('invalid_json',)
]
schema = StructType([
    StructField('name', StringType(), True),
    StructField('age', StringType(), True)
])
df = spark.createDataFrame(data, ['json_str'])

parsed_df = df.withColumn('parsed', from_json(col('json_str'), schema))

valid_df = parsed_df.filter(col('parsed').isNotNull())
error_df = parsed_df.filter(col('parsed').isNull())

error_count = error_df.count()

A 3

B 2

C 1

D 0

💡 Hint

Think about which JSON strings are invalid or cannot be parsed correctly.

from_json returns null only for malformed JSON. The first three records are valid JSON objects matching the StringType schema for both fields. The fourth record 'invalid_json' causes parsing to return null. Thus, error_df.count() is 1.

❓ data_output

intermediate

How many records are in the dead letter queue after applying this error handling logic?

Given a Spark DataFrame with a column 'data' containing JSON strings, you apply a schema to parse it. Records that fail parsing are sent to a dead letter queue. If the input data has 5 records and 2 of them are malformed JSON, how many records will be in the dead letter queue DataFrame?

A 3

B 2

C 5

D 0

💡 Hint

Dead letter queue contains only records that fail parsing.

Only the 2 malformed JSON records fail parsing and go to the dead letter queue. The other 3 parse successfully.

🔧 Debug

advanced

Identify the error in this Spark error handling code snippet

This PySpark code tries to parse JSON and separate valid and invalid records. What error will it raise when run?

Apache Spark

from pyspark.sql.functions import from_json, col

schema = 'name STRING, age INT'
df = spark.createDataFrame([('{"name":"Alice","age":30}',)], ['json_str'])
parsed_df = df.withColumn('parsed', from_json(col('json_str'), schema))
valid_df = parsed_df.filter(col('parsed').isNotNull())
error_df = parsed_df.filter(col('parsed').isNull())
error_count = error_df.count()

A No error, runs successfully

B SyntaxError: invalid syntax in schema string

C ValueError: cannot parse JSON string

D TypeError: schema must be a StructType

💡 Hint

Check the type of the schema argument passed to from_json.

The from_json function requires a StructType object for the schema, not a string. Passing a string causes a TypeError.

🚀 Application

advanced

Which option correctly implements a dead letter queue for malformed JSON records in Spark?

You want to parse a DataFrame column containing JSON strings. Malformed JSON records should be captured in a separate dead letter queue DataFrame. Which code snippet correctly implements this?

A

parsed_df = df.withColumn('parsed', from_json(col('json_str'), schema))
error_df = parsed_df.filter(col('parsed').isNull())

B

parsed_df = df.withColumn('parsed', from_json(col('json_str'), schema))
error_df = parsed_df.filter(col('json_str').isNull())

C

parsed_df = df.withColumn('parsed', from_json(col('json_str'), schema))
error_df = parsed_df.filter(col('json_str').isNotNull())

D

parsed_df = df.withColumn('parsed', from_json(col('json_str'), schema))
error_df = parsed_df.filter(col('parsed').isNotNull())

💡 Hint

Malformed JSON results in null in the parsed column.

The parsed column contains null for malformed JSON. Filtering where parsed is null captures bad records for the dead letter queue.

🧠 Conceptual

expert

Why is a dead letter queue important in Spark streaming error handling?

In Spark streaming pipelines, why do we use a dead letter queue for error handling instead of just dropping malformed records?

A To store malformed records for later analysis and reprocessing, ensuring data quality and traceability.

B To improve streaming speed by ignoring malformed records completely.

C To automatically fix malformed records using Spark's internal functions.

D To reduce storage costs by deleting bad data immediately.

💡 Hint

Think about data quality and auditing needs.

Dead letter queues keep bad records so teams can analyze and fix data issues later, improving data quality and traceability.

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