Unit testing Spark transformations in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Output of a Spark DataFrame filter transformation

What is the output of the following Spark code snippet?

Apache Spark

from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local').appName('test').getOrCreate()
data = [(1, 'apple'), (2, 'banana'), (3, 'carrot')]
df = spark.createDataFrame(data, ['id', 'fruit'])
filtered_df = df.filter(df.id > 1)
result = filtered_df.collect()

A [(Row(id=1, fruit='apple'))]

B [(Row(id=1, fruit='apple'), Row(id=2, fruit='banana'))]

C [(Row(id=2, fruit='banana'), Row(id=3, fruit='carrot'))]

D []

💡 Hint

Remember filter keeps rows where the condition is true.

The filter condition df.id > 1 keeps rows with id 2 and 3 only.

❓ data_output

intermediate

Result of a Spark DataFrame groupBy and count

What is the output of this Spark code after grouping and counting?

Apache Spark

from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local').appName('test').getOrCreate()
data = [('red', 1), ('blue', 2), ('red', 3), ('blue', 4), ('green', 5)]
df = spark.createDataFrame(data, ['color', 'value'])
grouped_df = df.groupBy('color').count().orderBy('color')
result = grouped_df.collect()

A [(Row(color='blue', count=2), Row(color='green', count=1), Row(color='red', count=2))]

B [(Row(color='red', count=3), Row(color='blue', count=2), Row(color='green', count=1))]

C [(Row(color='blue', count=1), Row(color='green', count=1), Row(color='red', count=1))]

D [(Row(color='red', count=2), Row(color='blue', count=3), Row(color='green', count=1))]

💡 Hint

Count how many times each color appears.

Colors 'red' and 'blue' appear twice each, 'green' once. Ordering is alphabetical by color.

🔧 Debug

advanced

Identify the error in Spark DataFrame join code

What error will this Spark code raise when executed?

Apache Spark

from pyspark.sql import SparkSession
spark = SparkSession.builder.master('local').appName('test').getOrCreate()
data1 = [(1, 'a'), (2, 'b')]
data2 = [(1, 'x'), (3, 'y')]
df1 = spark.createDataFrame(data1, ['id', 'val1'])
df2 = spark.createDataFrame(data2, ['id', 'val2'])
joined_df = df1.join(df2, on='ID')
result = joined_df.collect()

A AnalysisException: cannot resolve '`ID`' given input columns: [id, val1]

B No error, returns joined rows on id

C TypeError: join() missing required positional argument 'on'

D ValueError: join key must be a list or string

💡 Hint

Check column name case sensitivity in join keys.

The join key 'ID' does not match the actual column name 'id' (case sensitive), causing AnalysisException.

❓ visualization

advanced

Visualizing missing data counts in a Spark DataFrame

Which code snippet correctly computes the count of missing (null) values per column in a Spark DataFrame?

A df.select([F.count(F.when(F.col(c) == None, c)).alias(c) for c in df.columns])

B df.select([F.count(F.col(c).isNull()).alias(c) for c in df.columns])

C df.select([F.sum(F.col(c).isNull()).alias(c) for c in df.columns])

D df.select([F.count(F.when(F.col(c).isNull(), c)).alias(c) for c in df.columns])

💡 Hint

Use when() to filter nulls and count() to count them.

Option D uses when() to create a conditional column counting nulls, then counts them per column correctly.

🚀 Application

expert

Unit test for a Spark transformation function

Given a Spark transformation function that adds a new column 'double_value' doubling an existing 'value' column, which unit test code correctly verifies the transformation?

Apache Spark

def add_double_value_column(df):
    from pyspark.sql.functions import col
    return df.withColumn('double_value', col('value') * 2)

# Unit test code options below

A

def test_add_double_value_column(spark):
    data = [(1,), (2,), (3,)]
    df = spark.createDataFrame(data, ['value'])
    result_df = add_double_value_column(df)
    expected = [2, 4, 6]
    actual = [row.value for row in result_df.collect()]
    assert actual == expected

B

def test_add_double_value_column(spark):
    data = [(1,), (2,), (3,)]
    df = spark.createDataFrame(data, ['value'])
    result_df = add_double_value_column(df)
    expected = [2, 4, 6]
    actual = [row.double_value for row in result_df.collect()]
    assert actual == expected

C

def test_add_double_value_column(spark):
    data = [(1,), (2,), (3,)]
    df = spark.createDataFrame(data, ['value'])
    result_df = add_double_value_column(df)
    expected = [1, 2, 3]
    actual = [row.double_value for row in result_df.collect()]
    assert actual == expected

D

def test_add_double_value_column(spark):
    data = [(1,), (2,), (3,)]
    df = spark.createDataFrame(data, ['value'])
    result_df = add_double_value_column(df)
    expected = [2, 4, 6]
    actual = [row.double_value for row in df.collect()]
    assert actual == expected

💡 Hint

Check that the new column values are double the original values.

Option B collects the new column 'double_value' from the transformed DataFrame and compares it to the expected doubled values.

Practice

1. What is the main purpose of unit testing Spark transformations?

easy

A. To check small parts of Spark code for errors early

B. To improve the speed of Spark jobs

C. To deploy Spark applications automatically

D. To visualize Spark data outputs

Solution

1. Step 1: Understand unit testing purpose

   Unit testing focuses on checking small parts of code to find errors early.

2. Step 2: Relate to Spark transformations

   Testing Spark transformations means verifying each step works correctly before full runs.

3. Final Answer:

   To check small parts of Spark code for errors early -> Option A

4. Quick Check:

   Unit testing = Early error detection [OK]

Hint: Unit tests catch errors early in small code parts [OK]

Common Mistakes:

• Confusing unit testing with performance tuning
• Thinking unit tests deploy code
• Assuming unit tests create visualizations

2. Which of the following is the correct way to compare actual and expected DataFrames in a Spark unit test?

easy

A. print(actual_df) == print(expected_df)

B. assert actual_df.collect() == expected_df.collect()

C. actual_df.show() == expected_df.show()

D. actual_df.count() != expected_df.count()

Solution

1. Step 1: Understand DataFrame comparison

   Comparing DataFrames requires checking their collected data, not just printing or counts.

2. Step 2: Identify correct assertion

   Using assert with collected rows compares actual data content correctly.

3. Final Answer:

   assert actual_df.collect() == expected_df.collect() -> Option B

4. Quick Check:

   Compare collected data for equality [OK]

Hint: Use collect() to compare DataFrame contents in tests [OK]

Common Mistakes:

• Comparing printed output instead of data
• Using show() which returns None
• Comparing only row counts without data

3. Given the following Spark code in a unit test:

input_df = spark.createDataFrame([(1, 'a'), (2, 'b')], ['id', 'value'])
transformed_df = input_df.filter("id > 1")
result = transformed_df.collect()

What is the value of result?

medium

A. [Row(id=1, value='a')]

B. []

C. [Row(id=1, value='a'), Row(id=2, value='b')]

D. [Row(id=2, value='b')]

Solution

1. Step 1: Understand the filter condition

   The filter "id > 1" keeps rows where id is greater than 1, so only id=2 remains.

2. Step 2: Collect the filtered rows

   Collect returns a list of Row objects matching the filter, here only one row with id=2 and value='b'.

3. Final Answer:

   [Row(id=2, value='b')] -> Option D

4. Quick Check:

   Filter id > 1 keeps only id=2 [OK]

Hint: Filter keeps rows matching condition before collect [OK]

Common Mistakes:

• Including rows with id=1 mistakenly
• Confusing collect output format
• Assuming filter excludes id=2

4. You wrote this unit test code:

expected_df = spark.createDataFrame([(1, 'x')], ['id', 'value'])
actual_df = spark.createDataFrame([(1, 'x')], ['id', 'value'])
assert expected_df == actual_df

Why does this test fail?

medium

A. DataFrames cannot be compared directly with '=='

B. The schemas of DataFrames are different

C. The data in DataFrames is different

D. Spark session is not initialized

Solution

1. Step 1: Understand DataFrame equality

   Spark DataFrames do not support direct '==' comparison; it checks object identity, not content.

2. Step 2: Correct way to compare DataFrames

   Use methods like collect() or exceptAll() to compare data content, not '=='.

3. Final Answer:

   DataFrames cannot be compared directly with '==' -> Option A

4. Quick Check:

   Use collect() for DataFrame equality, not '==' [OK]

Hint: Never use '==' to compare DataFrames directly [OK]

Common Mistakes:

• Assuming '==' compares data content
• Ignoring schema differences
• Forgetting to initialize Spark session

5. You want to unit test a Spark transformation that removes duplicate rows based on the 'id' column, keeping the first occurrence. Which approach correctly tests this transformation?

hard

A. Create input DataFrame without duplicates, apply transformation, assert output equals input

B. Create input DataFrame with duplicates, apply transformation, assert output count equals input count

C. Create input DataFrame with duplicates, apply transformation, assert output equals expected DataFrame without duplicates

D. Create input DataFrame with duplicates, apply transformation, assert output is empty

Solution

1. Step 1: Prepare input with duplicates

   To test duplicate removal, input must contain duplicate rows on 'id'.

2. Step 2: Apply transformation and compare output

   After transformation, output should have duplicates removed; compare with expected DataFrame without duplicates.

3. Final Answer:

   Create input DataFrame with duplicates, apply transformation, assert output equals expected DataFrame without duplicates -> Option C

4. Quick Check:

   Test removes duplicates by comparing expected output [OK]

Hint: Test duplicate removal by comparing to expected no-duplicate DataFrame [OK]

Common Mistakes:

• Testing with no duplicates in input
• Comparing only row counts, not content
• Expecting empty output after removal