Integration testing pipelines in Apache Spark - Step-by-Step Execution

Concept Flow - Integration testing pipelines

Start: Define pipeline components

↓

Write integration tests

↓

Set up test data

↓

Run pipeline on test data

↓

Check outputs vs expected

↓

Pass?

No→Log errors & debug

Yes↓

Pipeline integration verified

↓

End

Integration testing pipelines means running the whole data flow with test data to check if all parts work together correctly.

Execution Sample

Apache Spark

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
input_df = spark.createDataFrame([(1, 'a'), (2, 'b')], ['id', 'value'])
result_df = pipeline_function(input_df)
expected_output = [(1, 'a_processed'), (2, 'b_processed')]
assert result_df.collect() == expected_output

This code runs a pipeline function on test data and checks if the output matches what we expect.

Execution Table

Step	Action	Input Data	Output Data	Check Result
1	Create test input DataFrame	[{id:1,value:'a'},{id:2,value:'b'}]	DataFrame with 2 rows	N/A
2	Run pipeline_function on input_df	DataFrame with 2 rows	Processed DataFrame	N/A
3	Collect output to list	Processed DataFrame	[Expected rows]	N/A
4	Compare output with expected_output	[Expected rows]	[Expected rows]	Pass if equal
5	Test ends	N/A	N/A	Pass - pipeline integration works

💡 Test ends after output matches expected results, confirming pipeline integration success.

Variable Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	Final
input_df	None	DataFrame with 2 rows	DataFrame with 2 rows	DataFrame with 2 rows	DataFrame with 2 rows
result_df	None	None	Processed DataFrame	Processed DataFrame	Processed DataFrame
output_list	None	None	None	[Expected rows]	[Expected rows]

Key Moments - 2 Insights

Why do we collect the DataFrame to a list before comparing?

Because Spark DataFrames are lazy and distributed, collecting converts them to local data for easy comparison, as shown in step 3 of the execution_table.

What if the output does not match expected_output?

The test fails and you log errors to debug the pipeline, as indicated by the 'No' branch after 'Pass?' in the concept_flow.

Visual Quiz - 3 Questions

Look at the execution_table, what is the output after step 2?

A List of expected rows

B Test input DataFrame

C Processed DataFrame

D None

💡 Hint

Check the 'Output Data' column in row 2 of the execution_table.

At which step do we verify if the pipeline output matches the expected output?

A Step 4

B Step 3

C Step 1

D Step 5

💡 Hint

Look at the 'Check Result' column in the execution_table for the step that says 'Pass if equal'.

If the test input had 3 rows instead of 2, how would the variable_tracker change after Step 1?

A result_df would have 3 rows

B input_df would have 3 rows

C output_list would be empty

D No change in input_df

💡 Hint

Variable_tracker shows input_df after Step 1 as the test input DataFrame size.

Concept Snapshot

Integration testing pipelines:
- Run full pipeline on test data
- Collect output locally
- Compare output with expected
- Pass means all parts work together
- Fail means debug pipeline steps

Full Transcript

Integration testing pipelines means running the entire data pipeline with prepared test data to check if all parts work together correctly. We start by defining the pipeline components and writing integration tests. Then we set up test data and run the pipeline on it. After running, we collect the output data locally and compare it with the expected results. If they match, the test passes and confirms the pipeline integration works. If not, we log errors and debug. The execution table shows step-by-step actions: creating test input, running the pipeline, collecting output, comparing results, and ending the test. Variables like input_df and result_df change as the pipeline runs. Collecting the DataFrame to a list is important because Spark DataFrames are lazy and distributed, so we need local data to compare easily. If output does not match expected, the test fails and debugging is needed. This process ensures the pipeline works end-to-end as expected.

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