Integration testing pipelines in Apache Spark

Introduction

Integration testing pipelines help check if different parts of a data process work well together. It ensures the whole data flow runs smoothly from start to end.

When you want to verify that data moves correctly through multiple Spark jobs.

Before deploying a new data pipeline to production to catch errors early.

When combining data from different sources and transformations to ensure correctness.

To test if output data matches expected results after all pipeline steps.

When updating parts of a pipeline and needing to confirm nothing breaks.

Syntax

Apache Spark

def test_pipeline():
    # Setup input data
    input_df = spark.createDataFrame([])

    # Run pipeline function
    result_df = run_pipeline(input_df)

    # Define expected output
    expected_df = spark.createDataFrame([])

    # Compare results
    assert result_df.collect() == expected_df.collect()

Use small sample data for quick tests.

Compare DataFrames by collecting or using DataFrame equality methods.

Examples

This tests a pipeline that capitalizes the 'value' column.

Apache Spark

def test_simple_pipeline():
    input_data = [(1, 'a'), (2, 'b')]
    input_df = spark.createDataFrame(input_data, ['id', 'value'])

    result_df = run_pipeline(input_df)

    expected_data = [(1, 'A'), (2, 'B')]
    expected_df = spark.createDataFrame(expected_data, ['id', 'value'])

    assert result_df.collect() == expected_df.collect()

This tests a pipeline that filters out rows with score less than 10.

Apache Spark

def test_pipeline_with_filter():
    input_data = [(1, 10), (2, 5), (3, 20)]
    input_df = spark.createDataFrame(input_data, ['id', 'score'])

    result_df = run_pipeline(input_df)

    expected_data = [(1, 10), (3, 20)]
    expected_df = spark.createDataFrame(expected_data, ['id', 'score'])

    assert result_df.collect() == expected_df.collect()

Sample Program

This program creates a simple Spark pipeline that capitalizes text. The test checks if the pipeline output matches the expected capitalized data.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql.functions import upper

spark = SparkSession.builder.master('local').appName('IntegrationTest').getOrCreate()

def run_pipeline(df):
    # Capitalize the 'value' column
    return df.withColumn('value', upper(df['value']))

def test_pipeline():
    input_data = [(1, 'apple'), (2, 'banana')]
    input_df = spark.createDataFrame(input_data, ['id', 'value'])

    result_df = run_pipeline(input_df)

    expected_data = [(1, 'APPLE'), (2, 'BANANA')]
    expected_df = spark.createDataFrame(expected_data, ['id', 'value'])

    assert result_df.collect() == expected_df.collect()
    print('Integration test passed!')

if __name__ == '__main__':
    test_pipeline()
    spark.stop()

Important Notes

Always stop the Spark session after tests to free resources.

Use small datasets to keep tests fast and simple.

Collecting DataFrames for comparison works well for small test data.

Summary

Integration testing pipelines checks if all parts work together correctly.

Use small sample data and compare actual vs expected outputs.

Run tests before deploying to catch errors early.