Integration testing pipelines in Apache Spark - Mini Project: Build & Apply

Integration Testing Pipelines

📖 Scenario: You work as a data engineer building data pipelines using Apache Spark. To ensure your pipelines work correctly, you want to write integration tests that check the data flow and transformations.Imagine you have a small dataset of sales records and you want to test a pipeline that filters and aggregates this data.

🎯 Goal: Build a simple Spark pipeline that filters sales data for a specific product, sums the sales amounts, and write integration tests to verify the pipeline's correctness.

📋 What You'll Learn

Create a Spark DataFrame with sales data

Set a filter condition for the product name

Write a pipeline that filters and sums sales

Print the final aggregated sales amount

💡 Why This Matters

🌍 Real World

Data engineers build pipelines that process and transform data. Integration testing ensures the entire pipeline works correctly end-to-end.

💼 Career

Knowing how to write and test Spark pipelines is essential for roles like data engineer, data analyst, and data scientist working with big data.

1

Create the sales DataFrame

Create a Spark DataFrame called sales_df with these exact rows: {'product': 'apple', 'amount': 10}, {'product': 'banana', 'amount': 5}, {'product': 'apple', 'amount': 15}.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('IntegrationTest').getOrCreate()
# Create sales_df DataFrame with the specified rows
# Your code here

Hint

Use spark.createDataFrame() with a list of dictionaries to create sales_df.

2

Set the product filter

Create a variable called filter_product and set it to the string 'apple'.

Apache Spark

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('IntegrationTest').getOrCreate()
sales_data = [
    {'product': 'apple', 'amount': 10},
    {'product': 'banana', 'amount': 5},
    {'product': 'apple', 'amount': 15}
]
sales_df = spark.createDataFrame(sales_data)
# Create filter_product variable with value 'apple'
# Your code here

Hint

Just assign the string 'apple' to filter_product.

3

Filter and aggregate the sales

Create a new DataFrame called filtered_df by filtering sales_df where the product column equals filter_product. Then create a variable called total_sales that sums the amount column of filtered_df.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum

spark = SparkSession.builder.appName('IntegrationTest').getOrCreate()
sales_data = [
    {'product': 'apple', 'amount': 10},
    {'product': 'banana', 'amount': 5},
    {'product': 'apple', 'amount': 15}
]
sales_df = spark.createDataFrame(sales_data)
filter_product = 'apple'
# Filter sales_df for product == filter_product and assign to filtered_df
# Sum the amount column of filtered_df and assign to total_sales
# Your code here

Hint

Use filter() on sales_df and sum() from pyspark.sql.functions to sum the amounts.

4

Print the total sales

Write a print statement to display the text 'Total sales for apple:' followed by the value of total_sales.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import sum

spark = SparkSession.builder.appName('IntegrationTest').getOrCreate()
sales_data = [
    {'product': 'apple', 'amount': 10},
    {'product': 'banana', 'amount': 5},
    {'product': 'apple', 'amount': 15}
]
sales_df = spark.createDataFrame(sales_data)
filter_product = 'apple'
filtered_df = sales_df.filter(sales_df.product == filter_product)
total_sales = filtered_df.select(sum('amount')).collect()[0][0]
# Print the total sales with the message
# Your code here

Hint

Use an f-string in the print statement to show the message and the total_sales value.

Practice

1. What is the main goal of integration testing pipelines in Apache Spark?

easy

A. To check if all parts of the data pipeline work together correctly

B. To test only individual functions separately

C. To deploy the pipeline to production without testing

D. To write documentation for the pipeline

Solution

1. Step 1: Understand integration testing purpose

   Integration testing checks if different parts of a system work together as expected.

2. Step 2: Apply to Spark pipelines

   In Spark, integration testing ensures all pipeline stages (data reading, transformation, writing) function together correctly.

3. Final Answer:

   To check if all parts of the data pipeline work together correctly -> Option A

4. Quick Check:

   Integration testing = check all parts together [OK]

Hint: Integration tests check combined parts, not single functions [OK]

Common Mistakes:

• Confusing integration testing with unit testing
• Skipping tests before deployment
• Thinking integration tests only check code style

2. Which of the following is the correct way to create a small sample DataFrame for integration testing in Spark using Python?

easy

A. spark.makeDataFrame([(1, 'a'), (2, 'b')], ['id', 'value'])

B. spark.createDataFrame([(1, 'a'), (2, 'b')], ['id', 'value'])

C. spark.newDataFrame([(1, 'a'), (2, 'b')], ['id', 'value'])

D. spark.DataFrame.create([(1, 'a'), (2, 'b')], ['id', 'value'])

Solution

1. Step 1: Recall Spark DataFrame creation syntax

   The correct method to create a DataFrame from a list in PySpark is spark.createDataFrame(data, schema).

2. Step 2: Match options with correct syntax

   Only spark.createDataFrame([(1, 'a'), (2, 'b')], ['id', 'value']) uses spark.createDataFrame correctly with data and column names.

3. Final Answer:

   spark.createDataFrame([(1, 'a'), (2, 'b')], ['id', 'value']) -> Option B

4. Quick Check:

   Use spark.createDataFrame() to make test data [OK]

Hint: Remember: spark.createDataFrame() is the correct method [OK]

Common Mistakes:

• Using non-existent methods like newDataFrame or makeDataFrame
• Calling createDataFrame on DataFrame class instead of spark session
• Passing schema incorrectly

3. Given the following Spark code for integration testing, what will be the output of result_df.collect()?

input_df = spark.createDataFrame([(1, 10), (2, 20)], ['id', 'value'])
result_df = input_df.withColumn('value_plus_one', input_df['value'] + 1)

medium

A. [(1, 10), (2, 20)]

B. [(1, 11), (2, 21)]

C. [(1, 10, 11), (2, 20, 21)]

D. Error: withColumn requires a string as first argument

Solution

1. Step 1: Understand withColumn usage

   withColumn adds a new column with the given name and values. Here, 'value_plus_one' is added as 'value' + 1.

2. Step 2: Calculate new column values

   For id=1, value=10, value_plus_one=11; for id=2, value=20, value_plus_one=21.

3. Final Answer:

   [(1, 10, 11), (2, 20, 21)] -> Option C

4. Quick Check:

   withColumn adds new column with value+1 [OK]

Hint: withColumn adds new column; check column names carefully [OK]

Common Mistakes:

• Thinking withColumn replaces existing columns
• Ignoring the new column added
• Confusing collect() output format

4. You wrote this integration test code but it fails with an error. What is the mistake?

test_df = spark.createDataFrame([(1, 'x'), (2, 'y')], ['id', 'value'])
result_df = test_df.filter(test_df['value'] > 1)
result_df.show()

medium

A. DataFrame creation syntax is wrong

B. Column name 'value' is incorrect

C. Missing import for filter function

D. Cannot compare string column with '>' operator in Spark

Solution

1. Step 1: Analyze filter condition

   The filter uses '>' to compare string column 'value' with 1. Spark does not support '>' between string and integer.

2. Step 2: Understand Spark comparisons

   For mixed types, cast the column e.g. F.col('value').cast('int') or ensure matching types, not direct '>' operator.

3. Final Answer:

   Cannot compare string column with '>' operator in Spark -> Option D

4. Quick Check:

   Compatible types required for '>' in Spark filters [OK]

Hint: Cast types before using comparison operators in Spark [OK]

Common Mistakes:

• Assuming '>' works between strings and numbers in Spark
• Misnaming columns
• Forgetting to import Spark functions

5. You want to write an integration test for a Spark pipeline that reads JSON data, filters records with age > 30, and writes the result. Which approach best ensures your test is reliable and fast?

hard

A. Use a small in-memory DataFrame with sample JSON data and compare output DataFrame to expected results

B. Run the full pipeline on the entire production dataset every time

C. Manually check output files after running the pipeline

D. Skip testing and fix bugs after deployment

Solution

1. Step 1: Consider test data size and speed

   Using a small in-memory DataFrame makes tests fast and easy to run repeatedly.

2. Step 2: Compare actual vs expected outputs

   Comparing DataFrames programmatically ensures correctness without manual checks.

3. Step 3: Avoid full production runs or skipping tests

   Running full production data is slow; skipping tests risks bugs in production.

4. Final Answer:

   Use a small in-memory DataFrame with sample JSON data and compare output DataFrame to expected results -> Option A

5. Quick Check:

   Small sample data + compare outputs = reliable fast tests [OK]

Hint: Test small samples and compare outputs, not full data [OK]

Common Mistakes:

• Testing on full production data every time
• Manual output verification
• Skipping tests before deployment