Data Pipeline Testing Strategies: Quality Checks, Great Expectations, and CDC Testing

Testing data pipelines differs fundamentally from testing application code. Data can be syntactically correct but semantically wrong; distributions shift silently; schema changes break downstream consumers without warning. This guide covers a comprehensive strategy: unit tests for transformation logic, Spark DataFrame testing, Great Expectations for automated data profiling, schema contract testing, CDC event validation, and Testcontainers integration.

The Data Testing Pyramid

1. Unit tests: Test individual transformation functions in isolation
2. Integration tests: Test pipeline components against real containerised databases
3. Data quality tests: Assert properties of actual data at rest or in motion
4. Contract tests: Verify that data producers meet consumer schema expectations
5. End-to-end tests: Run the full pipeline on a representative sample

Unit Testing Transformation Logic

# transformations.py
import re

def normalize_phone(phone: str):
    digits = re.sub(r'\D', '', phone or '')
    if len(digits) == 10:                        return f'+1{digits}'
    if len(digits) == 11 and digits[0] == '1':  return f'+{digits}'
    return None

def calculate_ltv(orders: list) -> float:
    return sum(o['amount'] for o in orders if o['status'] == 'completed')

# test_transformations.py
import pytest
from transformations import normalize_phone, calculate_ltv

class TestNormalizePhone:
    def test_10_digit(self):       assert normalize_phone('5551234567')   == '+15551234567'
    def test_11_digit(self):       assert normalize_phone('15551234567')  == '+15551234567'
    def test_formatted(self):      assert normalize_phone('(555) 123-4567') == '+15551234567'
    def test_invalid_none(self):   assert normalize_phone('555') is None
    def test_none_input(self):     assert normalize_phone(None) is None

class TestCalculateLTV:
    def test_only_completed(self):
        orders = [{'status':'completed','amount':100.0},{'status':'cancelled','amount':50.0}]
        assert calculate_ltv(orders) == 100.0
    def test_empty(self):          assert calculate_ltv([]) == 0.0

Unit Testing Spark Transformations

Use pytest with a local SparkSession.

# conftest.py
import pytest
from pyspark.sql import SparkSession

@pytest.fixture(scope='session')
def spark():
    return SparkSession.builder.master('local[2]') \
        .config('spark.sql.shuffle.partitions', '2').getOrCreate()

# test_spark_transforms.py
from pyspark.sql import functions as F
from my_transforms import deduplicate_events, enrich_with_user_tier

def test_deduplicate_events(spark):
    data = [('e1','u1','2026-01-01'), ('e1','u1','2026-01-01'), ('e2','u2','2026-01-02')]
    df = spark.createDataFrame(data, ['event_id','user_id','date'])
    result = deduplicate_events(df)
    assert result.count() == 2
    assert result.filter(F.col('event_id') == 'e1').count() == 1

def test_enrich_with_user_tier(spark):
    events = spark.createDataFrame([('e1','u1',100.0)], ['event_id','user_id','amount'])
    users  = spark.createDataFrame([('u1','gold')],      ['user_id','tier'])
    result = enrich_with_user_tier(events, users)
    assert result.filter(F.col('user_id')=='u1').first()['tier'] == 'gold'

Great Expectations for Data Profiling

Great Expectations (GX) provides a framework for defining and automating data quality checks.

import great_expectations as gx

context = gx.get_context()
datasource = context.sources.add_pandas_filesystem(
    name='orders_datasource', base_directory='./data/orders/')
asset = datasource.add_csv_asset(
    name='orders', batching_regex=r'orders_(?P<date>\d{8})\.csv')

suite = context.add_expectation_suite('orders_quality_suite')
validator = context.get_validator(
    batch_request=asset.build_batch_request(), expectation_suite=suite)

validator.expect_column_to_exist('order_id')
validator.expect_column_values_to_not_be_null('order_id')
validator.expect_column_values_to_be_unique('order_id')
validator.expect_column_values_to_be_in_set(
    'status', {'pending', 'completed', 'cancelled', 'refunded'})
validator.expect_column_values_to_be_between('amount', min_value=0, max_value=100000)
validator.expect_column_median_to_be_between('amount', min_value=20, max_value=500)
validator.save_expectation_suite()

checkpoint = context.add_or_update_checkpoint(name='orders_cp', validator=validator)
results = checkpoint.run()
assert results.success, f'Data quality check failed: {results}'

Schema Contract Testing

Schema contracts prevent breaking producer changes from silently corrupting consumers.

import jsonschema

ORDER_SCHEMA = {
    'type': 'object',
    'required': ['order_id', 'customer_id', 'amount', 'status'],
    'properties': {
        'order_id':    {'type': 'string',  'minLength': 1},
        'customer_id': {'type': 'integer', 'minimum': 1},
        'amount':      {'type': 'number',  'exclusiveMinimum': 0},
        'status':      {'type': 'string',
                        'enum': ['pending','completed','cancelled','refunded']},
    },
    'additionalProperties': True  # allow additive new fields
}

def validate_order(order: dict) -> None:
    jsonschema.validate(instance=order, schema=ORDER_SCHEMA)

# Test: valid order passes
validate_order({'order_id':'O1','customer_id':1,'amount':99.0,'status':'pending'})

# Test: missing field raises ValidationError
try:
    validate_order({'order_id':'O2','customer_id':1,'status':'pending'})
    assert False, 'Should have raised'
except jsonschema.ValidationError:
    pass  # expected

Testing CDC Pipelines

def test_debezium_insert_event():
    event = {'op': 'c', 'before': None,
             'after': {'order_id': 'ORD-001', 'amount': 99.99, 'status': 'pending'},
             'ts_ms': 1716854400000}
    assert event['op'] == 'c'
    assert event['before'] is None
    assert event['after']['order_id'] is not None

def test_debezium_update_event():
    event = {'op': 'u',
             'before': {'order_id': 'ORD-001', 'status': 'pending'},
             'after':  {'order_id': 'ORD-001', 'status': 'completed'},
             'ts_ms': 1716854460000}
    assert event['before']['status'] != event['after']['status']
    assert event['after']['status'] == 'completed'

def test_cdc_processor_applies_upsert():
    processor = CDCProcessor(target_table='fct_orders')
    events = [
        {'op': 'c', 'after':  {'order_id': 'O1', 'amount': 100.0}},
        {'op': 'u', 'before': {'order_id': 'O1', 'amount': 100.0},
                    'after':  {'order_id': 'O1', 'amount': 120.0}},
        {'op': 'd', 'before': {'order_id': 'O2', 'amount': 50.0}, 'after': None},
    ]
    result = processor.apply_events(events)
    assert result['O1']['amount'] == 120.0
    assert 'O2' not in result  # deleted

Integration Testing with Testcontainers

import pytest
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope='module')
def postgres():
    with PostgresContainer('postgres:15') as pg:
        yield pg

def test_order_loader_integration(postgres):
    from mylib.loaders import OrderLoader
    loader = OrderLoader(dsn=postgres.get_connection_url())
    loader.create_schema()
    loader.load([{'order_id': 'T1', 'amount': 99.0, 'status': 'completed'}])
    rows = loader.query("SELECT * FROM orders WHERE order_id = 'T1'")
    assert len(rows) == 1
    assert rows[0]['amount'] == 99.0

Production Data Quality Monitoring

def check_daily_order_count(date: str, df) -> None:
    count = df.filter(f"order_date = '{date}'").count()
    baseline_avg, baseline_std = get_historical_stats('daily_order_count', lookback_days=30)
    z_score = abs(count - baseline_avg) / (baseline_std or 1)
    if z_score > 3:
        raise DataQualityError(
            f'Order count {count} deviates {z_score:.1f} std devs from baseline')

# Null rate monitoring
for col in ['order_id', 'customer_id', 'amount']:
    null_pct = df.filter(f'{col} IS NULL').count() / df.count() * 100
    if null_pct > 0.1:
        alert(f'Column {col} has {null_pct:.2f}% nulls — threshold 0.1%')

Conclusion

A robust data pipeline testing strategy layers unit tests for transformation logic, Spark DataFrame tests with a local SparkSession, Great Expectations for automated data profiling and drift detection, JSON Schema contracts to enforce producer-consumer agreements, targeted CDC event tests, and Testcontainers-based integration tests. Combined with continuous data quality monitoring in production, these layers catch data issues before they silently corrupt business decisions.